From 1ab97b3972537bac1f2482f74b85e0ababeb8692 Mon Sep 17 00:00:00 2001 From: narawat Date: Sun, 2 Aug 2026 20:57:53 +0700 Subject: [PATCH] add yiemAgent --- etc.jl | 38 ++++++++---- src/type.jl | 167 +++++++++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 180 insertions(+), 25 deletions(-) diff --git a/etc.jl b/etc.jl index a35aae6..5d9b549 100644 --- a/etc.jl +++ b/etc.jl @@ -1,10 +1,28 @@ -# check if this column has vector embedding. if there is one, seach vector version instead - column_name_embedding = column_name * "_embedding" - if occursin(column_name_embedding, tables_schema[column_name_embedding]) - vector_column = Dict( - "table_name"=> table_name, - "column_name"=> column_name_embedding, - "operator"=> "vector_similarity", - "value"=> column_obj["value"] - ) - end \ No newline at end of file + + + +using Base.Threads + +println("Active Julia threads: ", nthreads()) + +# A CPU-heavy helper function +function compute_work(id, iterations) + println(" [Start] Task $id on Thread #", threadid()) + + total = 0.0 + for i in 1:iterations + total += sin(i) * cos(i) + end + + println(" [Done] Task $id on Thread #", threadid()) + return total +end + +# ==================================================================== +# 1. Basic @spawn and fetch +# ==================================================================== +println("\n--- 1. Single Task Spawning ---") + +# Threads.@spawn creates a Task and schedules it onto an available worker thread +task1 = Threads.@spawn compute_work("A", 10_000_000) +println(typeof(task1)) \ No newline at end of file diff --git a/src/type.jl b/src/type.jl index ef15777..c206fe4 100644 --- a/src/type.jl +++ b/src/type.jl @@ -1,5 +1,6 @@ module type - export agent, sommelier, companion, virtualcustomer, agentContext + export agent, sommelier, companion, virtualcustomer, agentContext, yiemAgent, + run_agent, take_response, follow_up, stop_agent using Dates, UUIDs, DataStructures, JSON, NATS @@ -168,6 +169,9 @@ end abstract type agent end +""" +docstring +""" mutable struct yiemAgent <: agent # High-level agent wrapper _state::agentState # Current state (prompt, model, messages, tools, etc.) @@ -176,13 +180,15 @@ mutable struct yiemAgent <: agent # High-level agent wrapper # if agent is running, it process user message after # the current tool call finished. - followUpQueue::Channel # Messages queued via followUp() during agent is + followUpQueue::Channel # Messages queued via follow_up() during agent is # running. After the agent loop process all input_ch - # and the agent isn't use tool call. it then process - # followUp message + # and the agent isn't using tool call, it processes + # followUp messages - output_ch::Channel # agent respond message to user after it process all - # user message in input_ch and all followUp message. + output_ch::Channel # agent sends response message to user after processing + # all user messages in input_ch and all followUp messages. + + _task::Union{Task, Nothing} # Background task running the agent loop formatMsgForLLM::Function # Convert agent messages to LLM message format preprocessMessages ::Union{Function, Nothing} # Preprocess/transform messages before sending to LLM @@ -190,35 +196,43 @@ mutable struct yiemAgent <: agent # High-level agent wrapper afterToolCall::Union{Function, Nothing} # Callback invoked after executing a tool call prepareNextTurn::Union{Function, Nothing} # Callback to prepare the next conversation turn prepareNextTurnWithContext::Union{Function, Nothing} # Same but receives context - activeRun::Union{Bool, Nothing} # tracks the currently executing agent run state sessionId::Union{String, Nothing} # Optional session identifier maxRetryDelayMs::Union{Int64, Nothing} # Maximum delay between retries (ms) toolExecution::toolExecutionMode # Default: run tool calls sequentially or in parallel end -# Outer constructor — clean keyword API +""" +docstring +""" function yiemAgent( ; systemPrompt::String="", - model::llmModel=llmModel{String}("", "", "unknown", "unknown", "", false, String[], modelCost(0.0, 0.0, 0.0, 0.0), 0, 0), + model=nothing, tools::Vector{agentTool}=agentTool[], messages::Vector{agentMessage}=agentMessage[], formatMsgForLLM::Function=defaultformatMsgForLLM, - preprocessMessages ::Union{Function, Nothing}=nothing, + preprocessMessages::Union{Function, Nothing}=nothing, beforeToolCall::Union{Function, Nothing}=nothing, afterToolCall::Union{Function, Nothing}=nothing, prepareNextTurn::Union{Function, Nothing}=nothing, prepareNextTurnWithContext::Union{Function, Nothing}=nothing, sessionId::Union{String, Nothing}=nothing, maxRetryDelayMs::Union{Int64, Nothing}=nothing, - toolExecution::toolExecutionMode=EXECUTION_PARALLEL, + toolExecution=nothing, ) - new( + # Create channels: input (user -> agent), followUp (async queue), output (agent -> user) + input_ch = Channel(16) + followUp = Channel(32) + output_ch = Channel(16) + + # Create struct with a placeholder task, then spawn and replace it + agent = yiemAgent( agentState(systemPrompt, model, tools, messages), - Channel(16), + input_ch, + followUp, + output_ch, + nothing, # placeholder — replaced below formatMsgForLLM, preprocessMessages, - onPayload, - onResponse, beforeToolCall, afterToolCall, prepareNextTurn, @@ -227,6 +241,129 @@ function yiemAgent( maxRetryDelayMs, toolExecution, ) + + # Spawn the background loop and attach it + agent._task = @spawn _agent_loop(agent) + + 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) + 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 + try + put!(agent.output_ch, assistantMessage( + role="assistant", + content=[textContent("Agent error: $(sprint(showerror, e))")], + api="", model="", usage=nothing, + stopReason="error", + errorMessage=strip(sprint(showerror, e)), + timestamp=now(), + )) + catch e2 + @error "Failed to send error response" error=e2 + end + 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) + # TODO: 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