ton/YiemAgent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity
Files
c13aeb3a74e593565b97b9d8f38a4f5a197a85ec
YiemAgent/src/agentCore.jl
T
ton 189bc2efcf update
2026-08-10 09:43:35 +07:00

1155 lines
38 KiB
Julia
Raw Blame History

module agentCore
export _agent_loop, OpenAiToUserMessage
using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization,
DataFrames, Base.Threads
using GeneralUtils
using ..type, ..utils
# ---------------------------------------------- 100 --------------------------------------------- #
"""
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(
assistantMsgCtx(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(
afterCtx(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
"""
emitToolExecutionEnd(finalized, emit)
Emits the `toolExecutionEnd` event with the finalized outcome,
signalling to listeners that the tool call has completed.
This event is part of the tool execution lifecycle:
`toolExecutionStart` → (zero or more `toolExecutionUpdate` events) →
`toolExecutionEnd`. Listeners (such as the TUI or logging systems)
use this lifecycle to track individual tool calls. The event carries
the final result so listeners have all the data they need without
requiring external state lookups.
# Arguments
- `finalized::finalizedOutcome`: The finalized outcome to report
- `emit::Function`: Event emitter
# Notes
- Part of a three-event lifecycle per tool call
- Carries the complete result so listeners need no external lookups
# Examples
```julia
# Emits a single event; returns nothing
emitToolExecutionEnd(finalized, emit)
# (emit receives toolExecEndEvent("call_1", "search_wine", result, false))
```
"""
function emitToolExecutionEnd(finalized::finalizedOutcome, emit::Function)
emit(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name,
finalized.result, finalized.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
emitToolExecutionEnd(finalized, emit)
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)
emitToolExecutionEnd(finalized, emit)
push!(entries, finalized)
else
task = task() do
executed = executePreparedToolCall(prep, signal, emit)
finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config, signal)
emitToolExecutionEnd(finalized, emit)
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:<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