ton/YiemAgent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity
Files
c7a98f1710b44d0eaeb1ee48b2d20b735467290c
YiemAgent/src/agentCore.jl
T
ton c7abf844ea update
2026-08-16 20:14:54 +07:00

1554 lines
60 KiB
Julia
Raw Blame History

module agentCore
export yiemAgent, _agentLoop, OpenAiToUserMessage, _extractToolCalls,
executePreparedToolCall, prepareToolCall, executeToolCallsSequential,
executeToolCallsParallel, executeToolCalls
using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization,
DataFrames, Base.Threads, NATS
using GeneralUtils
using ..type, ..utils, ..toolRegistry
function register_all_tools(store::toolRegistry.toolStore)
# Call parent module's version which has access to tool functions
parentmodule(@__MODULE__).register_all_tools(store)
end
# ---------------------------------------------- 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
_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 _processMessage()'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 # 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(
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=agentEventSink,
)
# Create channels: input (user -> agent), followUp (async queue), output (agent -> user)
inputChannel = Channel(16)
followUp = Channel(32)
outputChannel = Channel(16)
# load tools (statically registered at module init)
toolStore1 = toolStore(name="myagent")
register_all_tools(toolStore1)
# 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._agentLoop = @spawn _agentLoop(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 `_processMessage` 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 _agentLoop(agent::yiemAgent)
processMessageInputCh = Channel(32)
try
newUserMsg = nothing
processingTask = nothing
result = nothing
""" cases:
1) agent -> idle, user msg -> nothing
typeof(processingTask) == Nothing
agent.inputChannel -> nothing
agent.followUpChannel -> nothing
2) agent -> idle, user msg -> new msg
typeof(processingTask) == Nothing
agent.inputChannel -> new msg
agent.followUpChannel -> nothing
3) agent -> running, user msg -> nothing
typeof(processingTask) == Task, istaskdone(processingTask) -> false
agent.inputChannel -> nothing
agent.followUpChannel -> nothing
4) agent -> running, user msg -> new msg
typeof(processingTask) == Task, istaskdone(processingTask) -> false
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.inputChannel -> nothing
agent.followUpChannel -> new msg
6) agent -> idle, user msg -> nothing
typeof(processingTask) == Task, istaskdone(processingTask) -> true
agent.inputChannel -> nothing
agent.followUpChannel -> nothing
"""
while true
while newUserMsg === nothing
if isready(agent.inputChannel)
agent.agentEventSink("_agentLoop 1 agent._state.messages length $(length(agent._state.messages))")
# 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 agent._state.messages length $(length(agent._state.messages))")
# if agent runs is done but followUpChannel has messages,
# put new message in inputChannel instead
if isready(agent.followUpChannel)
agent.agentEventSink("_agentLoop 3 agent._state.messages length $(length(agent._state.messages))")
while isready(agent.followUpChannel)
followUpMsg = take!(agent.followUpChannel)
put!(agent.inputChannel, followUpMsg)
end
processingTask = nothing # reset
result = nothing # reset
else # _processMessage() done and no followUp message.
agent.agentEventSink("_agentLoop 4 agent._state.messages length $(length(agent._state.messages))")
result = deepcopy(agent._state.messages[end])
# filter out reasoningContent in-place
filter!(c -> !(c isa reasoningContent), result.content)
# format output
respondToUI = _assistantMessageToOpenAI(result)
put!(agent.outputChannel, respondToUI)
if !isempty(result.content) && result.content[1] isa textContent
agent.agentEventSink(result.content[1].text)
end
processingTask = nothing # reset
result = nothing # reset
end
end
yield()
end
end
# Check for shutdown signal
if newUserMsg === :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
newUserMsg = nothing # reset
break
else
# spawn new _processMessage() if it is not already running.
if processingTask === nothing
agent.agentEventSink("_agentLoop 5 agent._state.messages length $(length(agent._state.messages))")
# discard all messages in followUpChannel
while isready(agent.followUpChannel)
_ = take!(agent.followUpChannel)
end
# Dispatch message through the processing pipeline
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,
)
end
put!(processMessageInputCh, newUserMsg)
newUserMsg = 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 _processMessage(
inputChannel::Channel,
agentEventSink,
agentMsgHistory::Vector{agentMessage},
systemPrompt::String,
tools::OrderedDict{String, agentTool},
prepareContext::Function,
formatMessagesForLLM::Function,
llmCall,
beforeToolCall::Union{Function, Nothing},
afterToolCall::Union{Function, Nothing},
parallelToolExecute::Bool,
)::Nothing
agentEventSink("_processMessage 1 _state.messages length $(length(agentMsgHistory))")
# loop until llmCall() response didn't use tool calls
final_response = nothing
""" example message in 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")
),
]
)
"""
while true
# Drain inputChannel and convert OpenAI-format messages to userMessage type
while isready(inputChannel)
agentEventSink("_processMessage 2 _state.messages length $(length(agentMsgHistory))")
newUserMsg_openai = take!(inputChannel)
agentEventSink("_processMessage 3 _state.messages length $(length(agentMsgHistory))")
if newUserMsg_openai === :shutdown
agentEventSink("_processMessage 4 _state.messages length $(length(agentMsgHistory))")
# Re-emit shutdown signal for the loop to handle
put!(inputChannel, :shutdown)
break
end
agentEventSink("_processMessage 5 _state.messages length $(length(agentMsgHistory))")
newUserMsg = OpenAiToUserMessage(newUserMsg_openai)
push!(agentMsgHistory, newUserMsg)
agentEventSink("_processMessage 6 _state.messages length $(length(agentMsgHistory))")
end
agentEventSink("_processMessage 7 _state.messages length $(length(agentMsgHistory))")
# call prepareContext()
state = agentState(systemPrompt, nothing, tools, agentMsgHistory)
agentEventSink("_processMessage 8 _state.messages length $(length(agentMsgHistory))")
preparedContext = prepareContext(state, agentEventSink)
agentEventSink("_processMessage 9 _state.messages length $(length(agentMsgHistory))")
# Call formatMessagesForLLM() to format for LLM
formattedMessages = formatMessagesForLLM(preparedContext, agentEventSink)
agentEventSink("_processMessage 10 formattedMessages $formattedMessages")
""" response example
response = JSON.Object{String, Any}("finish_reason" => "tool_calls", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "", "reasoning_content" => "Here's a thinking process:\n\n1. **Identify User Request**: The user is asking for the weather in Bangkok.\n2. **Locate Relevant Tool**: I have a `getWeather` function available.\n3. **Check Function Parameters**:\n - `city` (required): City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\n - `units` (optional, default \"celsius\"): Temperature scale (\"celsius\" or \"fahrenheit\")\n4. **Prepare Parameters**:\n - `city`: \"Bangkok, Thailand\" (adding country for clarity, though just \"Bangkok\" might work, following the example format is safer)\n - `units`: Not specified, so I'll use the default (\"celsius\")\n5. **Execute Tool Call**: Call `getWeather` with `city: \"Bangkok, Thailand\"`\n6. **Anticipate Response**: The function will return current weather and forecast data for Bangkok. I'll then format it nicely for the user.\n - *Self-Correction/Verification during thought*: The prompt says \"city: City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\". I'll use \"Bangkok, Thailand\". The `units` parameter is optional, so I'll omit it to use the default.\n - Proceed with tool call.✅\n", "tool_calls" => Any[JSON.Object{String, Any}("type" => "function", "function" => JSON.Object{String, Any}("name" => "getWeather", "arguments" => "{\"city\":\"Bangkok, Thailand\"}"), "id" => "6fOilR5QPcdppbAAHluhkRyUDu3oWMAL")]))
"""
response = llmCall(formattedMessages)
agentEventSink(" llmCall " * string(response))
agentEventSink("_processMessage 11 _state.messages length $(length(agentMsgHistory))")
# Extract tool calls from LLM response content blocks
hasToolCalls, toolCallList, assistant_msg = _extractToolCalls(response)
agentEventSink("hasToolCalls: $hasToolCalls\ntoolCallList: $toolCallList _state.messages length $(length(agentMsgHistory))")
agentEventSink("assistant_msg " * string(assistant_msg))
agentEventSink("_processMessage 11-1")
# Add assistant message (tool calls or text) to history for next LLM turn
push!(agentMsgHistory, assistant_msg)
if hasToolCalls && length(toolCallList) > 0
# Build context and config for executeToolCalls
config = agentLoopConfig(
beforeToolCall,
afterToolCall,
parallelToolExecute ? "parallel" : "sequential",
)
signal = abortSignal(false)
agentEventSink("_processMessage 12 _state.messages length $(length(agentMsgHistory))")
# call executeToolCalls()
toolResultBatch = executeToolCalls(preparedContext, assistant_msg, toolCallList, config,
signal, agentEventSink)
agentEventSink("_processMessage 13 _state.messages length $(length(agentMsgHistory))")
# save toolResults to messages
for toolResult in toolResultBatch.messages
agentEventSink("toolResult " * string(toolResult))
push!(agentMsgHistory, toolResult)
end
agentEventSink("_processMessage 14 _state.messages length $(length(agentMsgHistory))")
if toolResultBatch.terminate
agentEventSink("_processMessage 15 _state.messages length $(length(agentMsgHistory))")
# If toolResultBatch requested termination, build a final response
final_content = [textContent("Tool execution completed.")]
for toolResult in toolResultBatch.messages
for content_block in toolResult.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
agentEventSink("_processMessage 16 _state.messages length $(length(agentMsgHistory))")
final_response = assistantMessage(
role="assistant",
content=final_content,
api=assistant_msg.api,
model=assistant_msg.model,
usage=assistant_msg.usage,
stopReason="tool_use_terminated",
errorMessage=if any(x -> x.isError, toolResultBatch.messages)
"One or more tool calls failed"
else
nothing
end,
timestamp=now(),
)
push!(agentMsgHistory, final_response)
break
end
else
agentEventSink("_processMessage 17 _state.messages length $(length(agentMsgHistory))")
# LLM did not use tool calls —
break
end
end
agentEventSink("_processMessage 18 _state.messages length $(length(agentMsgHistory))")
return nothing
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,
nothing, f.isError, now()
)
end
"""
_extractToolCalls(response) -> Tuple{Bool, Vector{agentToolCall}, Union{assistantMessageToolCall, assistantMessage}}
Extracts tool calls from the LLM response and constructs a message object.
Supports two response formats:
1. **Message format** (e.g. from LMStudio.jl / vLLM):
`response["message"]["tool_calls"]` — array of tool call objects with
`"type" => "function"`, `"function" => Dict("name" => ..., "arguments" => "...")`,
and `"id"`. The `"arguments"` value is a JSON string that gets parsed via `JSON.parse`.
2. **Content blocks format** (e.g. from OpenAI API):
`response.content` — array of content blocks. Blocks with `"type" => "tool_calls"`
contain a `"tool_calls"` array in format 1. Blocks with `"type" => "tool_call"`
have `"name"`, `"arguments"`, `"id"` at the top level (already-parsed args dict).
When `hasToolCalls` is true, returns an `assistantMessageToolCall` with the tool calls
and reasoning content. When `hasToolCalls` is false, returns an `assistantMessage`
with text/content blocks from the response.
The message is constructed from:
- `reasoning_content` (string) → stored in `reasoning` field (for tool calls) or `textContent` (for text)
- `response.content` blocks (text/reasoning) → added to content for text responses
- Top-level `api`, `provider`, `model`, `usage` → copied to the message
- `finish_reason` → used as `stopReason`
# Arguments
- `response`: LLM response object (Dict/JSON.Object or struct with `.content` field)
# Returns
- `Tuple{Bool, Vector{agentToolCall}, Union{assistantMessageToolCall, assistantMessage}}`:
`(hasToolCalls, toolCallList, message)` where message is `assistantMessageToolCall`
when tool calls exist, `assistantMessage` otherwise
# Example
1) llm_useTool = JSON.Object{String, Any}("finish_reason" => "tool_calls", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "", "reasoning_content" => "Here's a thinking process:\n\n1. **Identify User Request**: The user is asking for the weather in Bangkok.\n2. **Locate Relevant Tool**: I have a `getWeather` function available.\n3. **Check Function Parameters**:\n - `city` (required): City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\n - `units` (optional, default \"celsius\"): Temperature scale (\"celsius\" or \"fahrenheit\")\n4. **Prepare Parameters**:\n - `city`: \"Bangkok, Thailand\" (adding country for clarity, though just \"Bangkok\" might work, following the example format is safer)\n - `units`: Not specified, so I'll use the default (\"celsius\")\n5. **Execute Tool Call**: Call `getWeather` with `city: \"Bangkok, Thailand\"`\n6. **Anticipate Response**: The function will return current weather and forecast data for Bangkok. I'll then format it nicely for the user.\n - *Self-Correction/Verification during thought*: The prompt says \"city: City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\". I'll use \"Bangkok, Thailand\". The `units` parameter is optional, so I'll omit it to use the default.\n - Proceed with tool call.✅\n", "tool_calls" => Any[JSON.Object{String, Any}("type" => "function", "function" => JSON.Object{String, Any}("name" => "getWeather", "arguments" => "{\"city\":\"Bangkok, Thailand\"}"), "id" => "6fOilR5QPcdppbAAHluhkRyUDu3oWMAL")]))
2) llm_notUseTool = JSON.Object{String, Any}("finish_reason" => "stop", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "The weather in London, UK is currently Sunny with a temperature of 22°C.", "reasoning_content" => "The user asked for the weather in London.\nI called the `getWeather` tool for London, UK.\nThe response indicates it's Sunny and 22°C.\nI will convey this information to the user.\n"))
"""
function _extractToolCalls(response)
hasToolCalls = false
toolCallList = agentToolCall[]
# Helper: parse args (JSON string -> Dict, or pass through)
parse_args(raw) = raw isa AbstractDict && !(raw isa Dict{String,Any}) ?
Dict{String,Any}(raw) :
raw isa String ? JSON.parse(raw) :
raw isa Dict{String,Any} ? raw : Dict{String,Any}()
# Helper: build agentToolCall (positional)
make_tc(tc_data, default_id=string(uuid4())) = begin
func = get(tc_data, "function", Dict{String,Any}())
args = parse_args(get(func, "arguments", "{}"))
name = get(func, "name", "")
id_val = get(tc_data, "id", default_id)
agentToolCall("function", id_val, name, args)
end
# Format 1: response["message"]["tool_calls"] (LMStudio.jl / vLLM style)
msg = get(response, "message", nothing)
if msg !== nothing && msg isa AbstractDict
tc_array = get(msg, "tool_calls", nothing)
if tc_array !== nothing && tc_array isa Vector
for tc_data in tc_array
if tc_data isa AbstractDict
hasToolCalls = true
push!(toolCallList, make_tc(tc_data))
end
end
end
end
# Format 2: response.content blocks (OpenAI API style)
if !hasToolCalls
content = nothing
if response isa AbstractDict
content = get(response, "content", nothing)
else
try
content = getfield(response, :content)
catch
content = nothing
end
end
if content isa Vector
for content_block in content
if content_block isa AbstractDict
if get(content_block, "type", "") == "tool_calls"
for tc_data in get(content_block, "tool_calls", [])
if tc_data isa AbstractDict
hasToolCalls = true
push!(toolCallList, make_tc(tc_data))
end
end
elseif get(content_block, "type", "") == "tool_call"
hasToolCalls = true
tc = agentToolCall(
"function",
get(content_block, "id", string(uuid4())),
get(content_block, "name", ""),
get(content_block, "arguments", Dict{String,Any}()),
)
push!(toolCallList, tc)
end
end
end
end
end
# ── Construct assistantMessage from response ──────────────────────
finish_reason = get(response, "finish_reason", nothing)
stop_reason = finish_reason isa AbstractString ? String(finish_reason) : "end_turn"
api = get(response, "api", "")
provider = get(response, "provider", "")
model = get(response, "model", nothing)
usage = get(response, "usage", nothing)
# Collect reasoning from reasoning_content field (Format 1: Anthropic-style)
reasoning_text = get(response, "reasoning_content", nothing)
if reasoning_text === nothing && msg !== nothing && msg isa AbstractDict
reasoning_text = get(msg, "reasoning_content", nothing)
end
if reasoning_text isa String
reasoning_text = reasoning_text
elseif reasoning_text isa reasoningContent
reasoning_text = reasoning_text.text
else
reasoning_text = ""
end
reasoning_content = !isempty(reasoning_text) ? [reasoningContent(reasoning_text)] : reasoningContent[]
# Collect content blocks from response.content array (Format 2: OpenAI-style)
# or as a plain string (Format 1: LMStudio.jl non-tool-calls response)
content_from_response = get(response, "content", nothing)
content_blocks = Vector{messageContent}()
if content_from_response isa Vector
for block in content_from_response
if block isa AbstractDict
if get(block, "type", "") == "text"
push!(content_blocks, textContent(get(block, "text", "")))
elseif get(block, "type", "") == "tool_call"
# tool_call blocks — don't add text content for these
elseif get(block, "type", "") == "tool_calls"
# tool_calls blocks — don't add text content for these
elseif get(block, "type", "") == "reasoning"
push!(content_blocks, reasoningContent(get(block, "text", "")))
else
push!(content_blocks, textContent(get(block, "text", "")))
end
end
end
elseif content_from_response isa AbstractString && !isempty(content_from_response)
push!(content_blocks, textContent(content_from_response))
end
# Format 1 fallback: response["message"]["content"] as plain string
if isempty(content_blocks) && msg !== nothing && msg isa AbstractDict
msg_content = get(msg, "content", nothing)
if msg_content isa AbstractString && !isempty(msg_content)
push!(content_blocks, textContent(msg_content))
end
end
# Combine reasoning_content field + content array blocks, deduplicating reasoning
if !isempty(reasoning_content)
all_content = vcat(reasoning_content, content_blocks)
else
all_content = content_blocks
end
error_msg = get(response, "error_message", get(response, "errorMessage", nothing))
if usage === nothing || !(usage isa llmUsage)
usage = llmUsage(0, 0)
end
# Role: prefer Format 1 nested message, fallback to top-level, default "assistant"
role = get(response, "role", "assistant")
if role == "assistant" && msg !== nothing && msg isa AbstractDict
nested_role = get(msg, "role", nothing)
if nested_role isa AbstractString
role = nested_role
end
end
if hasToolCalls
assistant_msg = assistantMessageToolCall(
role = role,
toolCalls = toolCallList,
content = all_content,
api = api isa AbstractString ? String(api) : "",
provider = provider isa AbstractString ? String(provider) : "",
model = model,
usage = usage,
stopReason = stop_reason,
errorMessage = error_msg,
timestamp = now(),
)
else
assistant_msg = assistantMessage(
role = role,
content = all_content,
api = api isa AbstractString ? String(api) : "",
provider = provider isa AbstractString ? String(provider) : "",
model = model,
usage = usage,
stopReason = stop_reason,
errorMessage = error_msg,
timestamp = now(),
)
end
return hasToolCalls, toolCallList, assistant_msg
end
"""
shouldTerminate(finalizedCalls::Vector{finalizedOutcome}) -> Bool
The `terminate` flag is set by tool implementations, not by the agent
or the LLM. It signals that the tool itself has completed the user's
request or encountered a fatal condition, so the agent should stop
processing further turns without calling the LLM again.
Common scenarios where a tool sets `terminate: true`:
- **Task completion**: one-shot tools like `deploy`, `submit`, or
`send_payment` finish their work and report directly to the user
instead of asking the LLM "what next?"
- **Unrecoverable error**: a tool hits a fatal condition (database
connection lost, auth token expired) and stops the agent from
retrying endlessly.
- **Async handoff**: a tool triggers a long-running external operation
and wants the agent to stop now; the external system will resume
the agent later via `continue()`.
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` 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::assistantMessageToolCall,
toolCall::agentToolCall,
config::agentLoopConfig,
signal::abortSignal,
agentEventSink
)::Union{preparedToolCall,immediateOutcome}
agentEventSink("prepareToolCall 1")
tool = get(context.tools, toolCall.name, nothing) # pick a called tool from tool store
if tool === nothing
agentEventSink("prepareToolCall 2")
return immediateOutcome(createErrorToolResult("Tool $toolCall.name not found"), true)
end
try
agentEventSink("prepareToolCall 3")
# 1. prepare arguments (tool-specific transform)
prepared = prepareToolCallArguments(tool, toolCall)
agentEventSink("prepared " * string(prepared.arguments))
agentEventSink("prepareToolCall 4")
validatedArgs = validateToolArguments(tool, prepared)
agentEventSink("validatedArgs " * string(validatedArgs))
agentEventSink("prepareToolCall 5")
# 2. beforeToolCall hook — can block
if config.beforeToolCall !== nothing
agentEventSink("prepareToolCall 6")
before = config.beforeToolCall(
beforeToolCallContext(assistantMsg, toolCall, validatedArgs, context),
signal
)
agentEventSink("prepareToolCall 7")
if signal.aborted
agentEventSink("prepareToolCall 8")
return immediateOutcome(createErrorToolResult("Operation aborted"), true)
end
if before !== nothing && before.block
agentEventSink("prepareToolCall 9")
return immediateOutcome(
createErrorToolResult(get(before, :reason, "Tool execution was blocked")), true)
end
end
agentEventSink("prepareToolCall 10")
return preparedToolCall(tool, toolCall, validatedArgs)
catch e
bt = catch_backtrace()
errMsg = sprint() do io
showerror(io, e, bt)
println(io)
end
agentEventSink(errMsg)
return immediateOutcome(createErrorToolResult(sprint(showerror, e)), 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},
agentEventSink,
)::executedOutcome
agentEventSink("executePreparedToolCall 1")
agentEventSink("executePreparedToolCall 2")
agentEventSink("executePreparedToolCall 3")
try
result = prep.tool.execute(prep.toolCall.id, prep.args, signal, agentEventSink)
agentEventSink(result.content[1].text)
agentEventSink("executePreparedToolCall 4")
return executedOutcome(result, false)
catch e
bt = catch_backtrace()
errMsg = sprint() do io
showerror(io, e, bt)
println(io)
end
agentEventSink(errMsg)
return executedOutcome(createErrorToolResult(sprint(showerror, e)), 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::assistantMessageToolCall,
prep::preparedToolCall,
executed::executedOutcome,
config::agentLoopConfig,
signal::Union{Nothing,abortSignal},
agentEventSink
)::finalizedOutcome
agentEventSink("finalizeExecutedToolCall 1")
result = executed.result
isError = executed.isError
agentEventSink("finalizeExecutedToolCall 2")
if config.afterToolCall !== nothing
try
after = config.afterToolCall(
afterToolCallContext(assistantMsg, prep.toolCall, prep.args, result, isError, context),
signal
)
agentEventSink("finalizeExecutedToolCall 3")
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 e
bt = catch_backtrace()
errMsg = sprint() do io
showerror(io, e, bt)
println(io)
end
agentEventSink(errMsg)
result = createErrorToolResult(sprint(showerror, e))
isError = true
end
end
agentEventSink("finalizeExecutedToolCall 4")
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::assistantMessageToolCall,
toolCalls::Vector{agentToolCall},
config::agentLoopConfig,
signal::abortSignal,
agentEventSink,
)::agentToolCallBatch
agentEventSink("executeToolCallsSequential 1")
finalizedCalls = finalizedOutcome[]
messages = toolResultMessage[]
for tc in toolCalls
agentEventSink("start tool execute $(tc.id), $(tc.name), $(tc.arguments)")
prep = prepareToolCall(context, assistantMsg, tc, config, signal, agentEventSink)
agentEventSink("executeToolCallsSequential " * string(prep.args))
if prep isa immediateOutcome
agentEventSink("executeToolCallsSequential 2-1")
finalized = finalizedOutcome(tc, prep.result, prep.isError)
agentEventSink("executeToolCallsSequential 2-2")
else
agentEventSink("executeToolCallsSequential 3")
#XXX
executed = executePreparedToolCall(prep, signal, agentEventSink)
agentEventSink("executeToolCallsSequential 3-1")
finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config,
signal, agentEventSink)
agentEventSink("executeToolCallsSequential 3-2")
end
agentEventSink("executeToolCallsSequential 4")
agentEventSink("finalized $(finalized.toolCall.id), $(finalized.toolCall.name),
$(finalized.result), $(finalized.isError)")
push!(messages, createToolResultMessage(finalized))
push!(finalizedCalls, finalized)
agentEventSink("executeToolCallsSequential 5")
if signal !== nothing && signal.aborted
break
end
end
agentEventSink("executeToolCallsSequential 6")
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::assistantMessageToolCall,
toolCalls::Vector{agentToolCall},
config::agentLoopConfig,
signal::abortSignal,
agentEventSink,
)::agentToolCallBatch
entries = Union{finalizedOutcome,Task}[]
for tc in toolCalls
agentEventSink(toolExecStartEvent(tc.id, tc.name, tc.arguments))
prep = prepareToolCall(context, assistantMsg, tc, config, signal, agentEventSink)
if prep isa immediateOutcome
finalized = finalizedOutcome(tc, prep.result, prep.isError)
agentEventSink(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name,
finalized.result, finalized.isError))
push!(entries, finalized)
else
t = Task() do
executed = executePreparedToolCall(prep, signal, agentEventSink)
finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config, signal)
agentEventSink(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name,
finalized.result, finalized.isError))
return finalized
end
schedule(t)
push!(entries, t)
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::assistantMessageToolCall,
toolCalls::Vector{agentToolCall},
config::agentLoopConfig,
signal::abortSignal,
agentEventSink,
)::agentToolCallBatch
agentEventSink("_executeToolCalls 1")
hasSequential = false
for tc in toolCalls
t = get(context.tools, tc.name, nothing)
if t !== nothing && !t.parallelToolExecute
hasSequential = true
break
end
end
agentEventSink("_executeToolCalls 2")
if config.toolExecution == "sequential" || hasSequential
agentEventSink("_executeToolCalls 3")
return executeToolCallsSequential(context, assistantMsg, toolCalls, config, signal,
agentEventSink)
else
agentEventSink("_executeToolCalls 4")
return executeToolCallsParallel(context, assistantMsg, toolCalls, config, signal,
agentEventSink)
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:<mime>;base64,<data>`) 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
Reference in New Issue View Git Blame Copy Permalink