V0.8.0 async think loop #40

Merged
ton merged 50 commits from v0.8.0-async_think_loop into v0.8.0 2026-08-08 04:13:05 +00:00
7 changed files with 624 additions and 1480 deletions
Showing only changes of commit 92eee07168 - Show all commits
+5 -2
View File
@@ -16,8 +16,11 @@ module YiemAgent
include("llmfunction.jl")
using .llmfunction
include("interface.jl")
using .interface
include("core.jl")
using .core
include("api.jl")
using .api
# ---------------------------------------------- 100 --------------------------------------------- #
+329
View File
@@ -0,0 +1,329 @@
module api
export prompt
using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization,
DataFrames, Serde
using GeneralUtils
using ..type, ..util, ..llmfunction
# ---------------------------------------------- 100 --------------------------------------------- #
"""
Send a message to the agent's input channel.
Blocks if the input channel buffer is full (capacity 16 by default).
"""
function run_agent(agent::yiemAgent, msg)
put!(agent.input_ch, msg)
return agent
end
"""
Take a response from the agent's output channel.
Blocks until the agent sends a response.
"""
function take_response(agent::yiemAgent)
return take!(agent.output_ch)
end
"""
Send a follow-up message while the agent is still processing.
Follow-up messages are processed after all input_ch messages
and before any tool call results are sent.
"""
function follow_up(agent::yiemAgent, msg)
put!(agent.followUpQueue, msg)
return agent
end
"""
Gracefully stop the agent.
Sends a :shutdown signal, waits for the task to finish, then closes all channels.
"""
function stop_agent(agent::yiemAgent)
put!(agent.input_ch, :shutdown)
try
fetch(agent._task)
catch e
if e isa TaskFailedException
rethrow(e)
end
end
close(agent.input_ch)
close(agent.output_ch)
close(agent.followUpQueue)
return nothing
end
""" Recursively convert dictionary-like variable (e.g. JSON.Object) into an OrderedDict.
The function walks any nested structure composed of `AbstractDict` (e.g., `JSON.Object`,
`Dict`, `OrderedDict`) and `AbstractArray` and produces a new tree where
every dictionary-like node is an `OrderedDict` and every array-like node is a `Vector{Any}`.
Scalar values (numbers, strings, booleans, `nothing`, etc.) are returned unchanged.
Does **not** mutate the input; it always allocates new containers.
# Arguments
- `x`
Any Julia value. If `x` is an `AbstractDict` it will be converted to an `OrderedDict`;
if it is an `AbstractArray` its elements will be processed recursively.
# Keyword Arguments
- `keytype::Type=Any`
The key type for the output OrderedDict. Use `String` for `OrderedDict{String,Any}`,
`Symbol` for `OrderedDict{Symbol,Any}`, or `Any` to preserve original key types.
- `sort_order::Union{Nothing, Vector}=nothing`
Vector of keys specifying the desired order. Keys are arranged in the specified order
first, followed by any remaining keys.
# Return
- A newly allocated nested structure composed of `OrderedDict{keytype,Any}` and `Vector{Any}`
that mirrors the input shape but uses ordered Julia containers.
# Notes
- The function treats any `AbstractDict` as a mapping source, so it works with
`JSON.Object`, `Dict`, `OrderedDict`, etc.
- Arrays are returned as `Vector{Any}` with their elements processed recursively.
# Examples
```jldoctest
julia> using JSON, DataStructures
julia> d = Dict(
"a" => 4,
"b" => 6,
"c" => Dict(
"d"=>7,
:e=>Dict(
"f"=>"hey",
"g"=>Dict(
"world"=>[1, "2", 3, Dict(:dd=>4.7)]
)
)
)
)
julia> jsonstring = JSON.json(d)
julia> A1 = JSON.parse(jsonstring) # A1 type is JSON.Object
julia> A2 = dictify(A1; keytype=String)
OrderedDict{String,Any} with 3 entries:
"a" => 4
"b" => 6
"c" => OrderedDict("d"=>7, "e"=>Dict("f"=>"hey", "g"=>Dict("world"=>[1, "2", 3, 4.7])))
julia> A3 = dictify(A1; keytype=Symbol)
OrderedDict{Symbol,Any} with 3 entries:
:a => 4
:b => 6
:c => OrderedDict(:d=>7, :e=>Dict("f"=>"hey", "g"=>Dict("world"=>[1, "2", 3, 4.7])))
julia> B1 = dictify(d; keytype=String)
OrderedDict{String, Any} with 3 entries:
```
**With sort_order:**
```jldoctest
julia> d = Dict("a"=>1, "b"=>2, "c"=>3)
julia> dictify(d; sort_order=["c", "a"])
OrderedDict{String,Int} with 3 entries:
"c" => 3
"a" => 1
"b" => 2
```
"""
function dictify(x::T; keytype::Type=Any, sort_order::Union{Nothing, Vector}=nothing
)::OrderedDict where {T<:AbstractDict}
# this function is example
end
end # module interface
+147
View File
@@ -0,0 +1,147 @@
module core
# export prompt
using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization,
DataFrames, Serde
using GeneralUtils
using ..type, ..util, ..llmfunction
# ---------------------------------------------- 100 --------------------------------------------- #
"""
Private agent loop. Runs in a background @task.
Waits on input_ch and followUpQueue, processing whichever has a message first.
"""
function _agent_loop(agent::yiemAgent)
try
while true
# Wait on either channel — the one with a message fires first
# Wait on either channel — the one with a message is taken first
msg = nothing
while msg === nothing
if isready(agent.input_ch)
msg = take!(agent.input_ch)
elseif isready(agent.followUpQueue)
msg = take!(agent.followUpQueue)
else
yield()
end
end
# Check for shutdown signal
if msg === :shutdown
break
end
# Dispatch message through the processing pipeline
result = _process_message(agent, msg)
# Send response to user
put!(agent.output_ch, result)
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 where you add your LLM call, tool execution, etc.
"""
function _process_message(agent::yiemAgent, msg)
# WORKING Replace with actual processing logic
#
# 1. Add msg to agent._state.messages
# 2. Call agent.formatMsgForLLM(agent._state) to format for LLM
# 3. If preprocessMessages is set, call agent.preprocessMessages(...)
# 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
# Placeholder: echo back the message as a simple response
@warn "TODO: implement _process_message"
return assistantMessage(
role="assistant",
content=[textContent("Received: $(msg)")],
api="", model="", usage=nothing,
stopReason="end_turn",
errorMessage=nothing,
timestamp=now(),
)
end
end # end of module
-1363
View File
File diff suppressed because it is too large Load Diff
+143 -115
View File
@@ -31,6 +31,23 @@ struct userMessage <: agentMessage # Message from the user
timestamp::Timestamp # When the message was sent
end
"""
Create a new user message.
# Arguments
- `role::String`: Always "user"
- `content::Vector{messageContent}`: Text and/or image content
- `timestamp::Timestamp`: When the message was sent
# Returns
- A new `userMessage` instance
# Examples
```julia
julia> msg = userMessage(content=[textContent("Hello")])
userMessage("user", [textContent("Hello")], DateTime(...))
```
"""
function userMessage(; role="user", content=Vector{messageContent}(), timestamp=now())
return userMessage(role, content, timestamp)
end
@@ -47,6 +64,29 @@ struct assistantMessage <: agentMessage # Message from the AI assistant
timestamp::Timestamp # When the message was received
end
"""
Create a new assistant message.
# Arguments
- `role::String`: Always "assistant"
- `content::Vector{messageContent}`: Text and/or image content
- `api::String`: API name used (e.g., "openai")
- `provider::String`: Provider name (e.g., "anthropic")
- `model::String`: Model identifier
- `usage::Usage`: Token usage for this message
- `stopReason::String`: Why generation stopped (e.g., "end_turn")
- `errorMessage::Union{String, Nothing}`: Error if generation failed
- `timestamp::Timestamp`: When the message was received
# Returns
- A new `assistantMessage` instance
# Examples
```julia
julia> msg = assistantMessage(content=[textContent("Hello!")], model="gpt-4")
assistantMessage("assistant", [textContent("Hello!")], "", "", "gpt-4", ..., "end_turn", nothing, DateTime(...))
```
"""
function assistantMessage(; role="assistant", content=Vector{messageContent}(),
api="", provider="", model="", usage=Usage(0, 0), stopReason="end_turn",
errorMessage=nothing, timestamp=now())
@@ -65,6 +105,29 @@ struct toolResultMessage <: agentMessage # Result returned from a tool execut
timestamp::Timestamp # When the result was recorded
end
"""
Create a new tool result message.
# Arguments
- `role::String`: Always "tool"
- `toolCallId::String`: ID matching the tool call
- `toolName::String`: Name of the executed tool
- `content::Vector{messageContent}`: Tool output content
- `details::Any`: Additional tool-specific details
- `usage::Union{Usage, Nothing}`: Token usage if applicable
- `addedToolNames::Union{Vector{String}, Nothing}`: Tools added during execution
- `isError::Bool`: Whether the tool call resulted in an error
- `timestamp::Timestamp`: When the result was recorded
# Returns
- A new `toolResultMessage` instance
# Examples
```julia
julia> msg = toolResultMessage(toolCallId="call_123", toolName="search", content=[textContent("results")])
toolResultMessage("tool", "call_123", "search", [textContent("results")], nothing, nothing, nothing, false, DateTime(...))
```
"""
function toolResultMessage(; role="tool", toolCallId="", toolName="",
content=Vector{messageContent}(), details=nothing, usage=nothing,
addedToolNames=nothing, isError=false, timestamp=now())
@@ -82,6 +145,21 @@ struct textContent <: messageContent # Plain text message content
text::String # The text content
end
"""
Create plain text message content.
# Arguments
- `text::String`: The text content
# Returns
- A new `textContent` instance
# Examples
```julia
julia> content = textContent("Hello, world!")
textContent("Hello, world!")
```
"""
function textContent(; text="")
return textContent(text)
end
@@ -91,6 +169,22 @@ struct imageContent <: messageContent # Image message content
mimeType::String # MIME type (e.g., "image/png")
end
"""
Create image message content.
# Arguments
- `data::String`: Base64-encoded image data
- `mimeType::String`: MIME type (e.g., "image/png")
# Returns
- A new `imageContent` instance
# Examples
```julia
julia> img = imageContent(data="base64data...", mimeType="image/png")
imageContent("base64data...", "image/png")
```
"""
function imageContent(; data="", mimeType="")
return imageContent(data, mimeType)
end
@@ -135,6 +229,27 @@ mutable struct agentState # Mutable runtime state of an agen
errorMessage::Union{String, Nothing} # Last error message
end
"""
Create a new mutable agent state.
Creates a deep copy of the provided tools and messages to isolate the
new state from external references.
# Arguments
- `systemPrompt::String`: System prompt text
- `model::llmModel`: LLM model to use (defaults to an unknown model)
- `tools::Vector{agentTool}`: Available tools (deep copied)
- `messages::Vector{agentMessage}`: Conversation messages (deep copied)
# Returns
- A new `agentState` instance with an empty pending tool calls list and no error
# Examples
```julia
julia> state = agentState(systemPrompt="You are a helpful assistant")
agentState("You are a helpful assistant", ..., agentTool[], agentMessage[], String[], nothing)
```
"""
function agentState(
systemPrompt::String="",
model::llmModel=llmModel{String}("", "", "unknown", "unknown", "", false, String[], modelCost(0.0, 0.0, 0.0, 0.0), 0, 0),
@@ -237,7 +352,34 @@ mutable struct yiemAgent <: agent # High-level agent wrapper
end
"""
docstring
Create a new yiemAgent instance with a background loop task.
Spawns a background `@spawn` task that runs the agent loop, listening
on `input_ch` and `followUpQueue` channels concurrently.
# Keyword Arguments
- `systemPrompt::String`: System prompt for the agent
- `model`: LLM model to use
- `tools::Vector{agentTool}`: Available tools (default: empty)
- `messages::Vector{agentMessage}`: Initial conversation messages (default: empty)
- `formatMsgForLLM::Function`: Convert agent messages to LLM message format (default: `defaultformatMsgForLLM`)
- `preprocessMessages::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`)
- `toolExecution`: Default tool execution mode (sequential or parallel) (default: `nothing`)
# Returns
- A new `yiemAgent` instance with an active background task
# Examples
```julia
julia> agent = yiemAgent(systemPrompt="You are a helpful assistant", model=my_model)
yiemAgent(agentState(...), Channel(...), Channel(...), Channel(...), ..., ...)
```
"""
function yiemAgent(
; systemPrompt::String="",
@@ -283,120 +425,6 @@ function yiemAgent(
return agent
end
# ============================================================================
# Agent loop — runs in background, processes messages from input_ch / followUp
# ============================================================================
"""
Private agent loop. Runs in a background @task.
Waits on input_ch and followUpQueue concurrently via select().
"""
function _agent_loop(agent::yiemAgent) #WORKING
try
while true
# Wait on either channel — the one with a message fires first
msg = select(agent.input_ch, agent.followUpQueue).val
# Check for shutdown signal
if msg === :shutdown
break
end
# Dispatch message through the processing pipeline
result = _process_message(agent, msg)
# Send response to user
put!(agent.output_ch, result)
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 where you add your LLM call, tool execution, etc.
"""
function _process_message(agent::yiemAgent, msg)
# PENDING Replace with actual processing logic
#
# 1. Add msg to agent._state.messages
# 2. Call agent.formatMsgForLLM(agent._state) to format for LLM
# 3. If preprocessMessages is set, call agent.preprocessMessages(...)
# 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
# Placeholder: echo back the message as a simple response
@warn "TODO: implement _process_message"
return assistantMessage(
role="assistant",
content=[textContent("Received: $(msg)")],
api="", model="", usage=nothing,
stopReason="end_turn",
errorMessage=nothing,
timestamp=now(),
)
end
# ============================================================================
# Public API — interaction helpers
# ============================================================================
"""
Send a message to the agent's input channel.
Blocks if the input channel buffer is full (capacity 16 by default).
"""
function run_agent(agent::yiemAgent, msg)
put!(agent.input_ch, msg)
return agent
end
"""
Take a response from the agent's output channel.
Blocks until the agent sends a response.
"""
function take_response(agent::yiemAgent)
return take!(agent.output_ch)
end
"""
Send a follow-up message while the agent is still processing.
Follow-up messages are processed after all input_ch messages
and before any tool call results are sent.
"""
function follow_up(agent::yiemAgent, msg)
put!(agent.followUpQueue, msg)
return agent
end
"""
Gracefully stop the agent.
Sends a :shutdown signal, waits for the task to finish, then closes all channels.
"""
function stop_agent(agent::yiemAgent)
put!(agent.input_ch, :shutdown)
try
fetch(agent._task)
catch e
if e isa TaskFailedException
rethrow(e)
end
end
close(agent.input_ch)
close(agent.output_ch)
close(agent.followUpQueue)
return nothing
end