From 4e592173a655467dc1ed04804984c241a7df2ba1 Mon Sep 17 00:00:00 2001 From: narawat Date: Wed, 12 Aug 2026 14:37:11 +0700 Subject: [PATCH] update --- etc.jl | 79 +++++++++++- src/agentCore.jl | 26 ++-- test/runtest.jl | 327 ----------------------------------------------- 3 files changed, 95 insertions(+), 337 deletions(-) diff --git a/etc.jl b/etc.jl index 4ce0950..7d92ff2 100644 --- a/etc.jl +++ b/etc.jl @@ -1,2 +1,77 @@ -# ── executeToolCalls() Julia pseudo code ────────────────────────── -# Full call stack from runLoop → executeToolCalls → prepare → execute → finalize → emit +i am not sure that's the case. see my NATS message log: + +┌ Info: debug +└ payload = "new user msg" +┌ Info: debug +└ payload = "new user msg" +┌ Info: debug +└ payload = "new user msg" +┌ Info: debug +└ payload = "new user msg" +┌ Info: debug +└ payload = "new user msg" +┌ Info: debug +└ payload = "new user msg" +┌ Info: debug +└ payload = "new user msg" +┌ Info: debug +└ payload = "_process_message 3" +┌ Info: debug +└ payload = "_process_message 5" +┌ Info: debug +└ payload = "_process_message 6" +┌ Info: debug +└ payload = "_process_message 7" + + +my NATS receiver report the following for a long time +┌ Info: debug +└ payload = "new user msg" + +untill I Ctrl + d so shutdown the process then i got the following report +┌ Info: debug +└ payload = "_process_message 3" +┌ Info: debug +└ payload = "_process_message 5" +┌ Info: debug +└ payload = "_process_message 6" +┌ Info: debug +└ payload = "_process_message 7" + + + + + + + + + +my point is if _process_message() actually run then this code in _process_message() +"raw_msg = take!(agent.inputChannel)" +should take the new msg message out of agent.inputChannel and there should be only one debug message showing +┌ Info: debug +└ payload = "new user msg" + +before reaching error("debug marker") + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/agentCore.jl b/src/agentCore.jl index e913e49..1526ce2 100644 --- a/src/agentCore.jl +++ b/src/agentCore.jl @@ -54,7 +54,7 @@ mutable struct yiemAgent <: agent # High-level agent wrapper sessionId::Union{String, Nothing} # Optional session identifier maxRetryDelayMs::Union{Int64, Nothing} # Maximum delay between retries (ms) parallelToolExecute::Bool # Default: false - agentEventSink::Function # agent emits its status via this function + agentEventSink # agent emits its status via this function end """ @@ -99,7 +99,7 @@ function yiemAgent( sessionId::Union{String, Nothing}=nothing, maxRetryDelayMs::Union{Int64, Nothing}=nothing, parallelToolExecute::Bool=false, - agentEventSink::Function=agentEventSink, + agentEventSink=agentEventSink, ) # Create channels: input (user -> agent), followUp (async queue), output (agent -> user) inputChannel = Channel(16) @@ -209,7 +209,8 @@ function _agent_loop(agent::yiemAgent) if isready(agent.inputChannel) # message will be taken in _process_message() - msg = fetch!(agent.inputChannel) + msg = fetch(agent.inputChannel) + agent.agentEventSink("new user msg") else yield() end @@ -235,9 +236,11 @@ function _agent_loop(agent::yiemAgent) # start _process_message loop if agent._state.activeRun == false + agent.agentEventSink("_agent_loop 2") # Dispatch message through the processing pipeline - processingTask = Threads.@spawn _process_message(agent) + processingTask = Threads.@spawn _process_message(agent) agent._state.activeRun = true + agent.agentEventSink("_agent_loop 3") end # during agent runs, check followUp message after _process_message() is done @@ -301,6 +304,7 @@ julia> # Currently returns a placeholder echo response ``` """ function _process_message(agent::yiemAgent)::assistantMessage + agent.agentEventSink("_process_message 1") # loop until llmCall() response didn't use tool calls final_response = nothing while true @@ -313,21 +317,26 @@ function _process_message(agent::yiemAgent)::assistantMessage 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) + agent.agentEventSink("_process_message 2") raw_msg = take!(agent.inputChannel) + agent.agentEventSink("_process_message 3") if raw_msg === :shutdown + agent.agentEventSink("_process_message 4") # Re-emit shutdown signal for the loop to handle put!(agent.inputChannel, :shutdown) break end + agent.agentEventSink("_process_message 5") user_msg = OpenAiToUserMessage(raw_msg) push!(agent._state.messages, user_msg) + agent.agentEventSink("_process_message 6") end # call agent.prepareContext() @@ -336,10 +345,11 @@ function _process_message(agent::yiemAgent)::assistantMessage # Call agent.formatMsgForLLM(agent._state) to format for LLM formatted_messages = agent.formatMsgForLLM(preparedContext) + agent.agentEventSink("_process_message 7") # Call llmCall() (blocking — the task waits here) + error("debug marker") response = agent.llmCall(formatted_messages) - - error(5555555) + agent.agentEventSink("_process_message 8") #WORKING Check if LLM used tool calls (inspect content for tool_call blocks) has_tool_calls = false diff --git a/test/runtest.jl b/test/runtest.jl index 18c4353..e69de29 100644 --- a/test/runtest.jl +++ b/test/runtest.jl @@ -1,327 +0,0 @@ -using JSON, Dates, UUIDs, PrettyPrinting, LibPQ, Base64, DataFrames, DataStructures, HTTP, Base64, - NATS, Base.Threads -using YiemAgent, GeneralUtils - -struct text2textInstructLLM - natsConn::NATS.Connection - topic::String - senderID::String - fileserver_url::String -end - -function (t::text2textInstructLLM)(openai_msg::Dict{String, Any}) - payloads = [("msg", openai_msg, "dictionary")] # List of tuples - _, msg_envelope_json_str = msghandler.smartpack( - t.topic, - payloads; - sender_id=t.senderID, - msg_purpose="text2text", - fileserver_url=t.fileserver_url) - - reply = NATS.request(t.natsConn, t.topic, msg_envelope_json_str, timeout=180) - - incoming_env_json_str = String(reply.payload) - incoming_env = msghandler.smartunpack(incoming_env_json_str) - _llm_response = incoming_env["payloads"][1][2] - llm_response = _llm_response["choices"][1]["message"]["content"] - return llm_response -end - - -# function get_embedding(text::AbstractArray{String}) -# documents_dict = Dict("documents" => text) -# payloads = [("documents", documents_dict, "dictionary")] -# _, msg_envelope_json_str = msghandler.smartpack( -# config["externalservice"]["servicesloadbalancer"]["nats"], -# payloads; -# msg_purpose="embedding", -# broker_url=config["nats_server_info"]["url"], -# fileserver_url=config["externalservice"]["fileserver"]["url"]) - -# reply = NATS.request(agent_conn, -# config["externalservice"]["servicesloadbalancer"]["nats"], -# msg_envelope_json_str, timeout=120) -# incoming_env_json_str = String(reply.payload) -# incoming_env = msghandler.smartunpack(incoming_env_json_str) -# embedding_response = incoming_env["payloads"][1][2] - -# return embedding_response -# end - - -# """ sql = "SELECT * FROM wine;" -# result = execute_sql_winedb(sql) -# """ -# function execute_sql_winedb(sql::T) where {T<:AbstractString} -# host_url, _port = split(config["externalservice"]["sommpanion_db"]["url"], ':') -# port = parse(Int, _port) -# dbname = "winedb" -# user = config["externalservice"]["sommpanion_db"]["user"] -# password = config["externalservice"]["sommpanion_db"]["password"] -# db_connection = LibPQ.Connection("host=$host_url port=$port dbname=$dbname user=$user password=$password") -# result = nothing -# try -# result = LibPQ.execute(db_connection, sql) -# catch e -# LibPQ.close(db_connection) -# end - -# LibPQ.close(db_connection) -# return result -# end - -# """ find similar sql from vector database -# sql = "SELECT * FROM wine;" -# result, distance = similar_sql_vectordb(sql) -# """ -# function similar_sql_vectordb(sql::T; maxdistance::Number=1) where {T<:AbstractString} -# tablename = "sqlllm_decision_repository" -# # get embedding of the query -# df = find_similar_text_from_vectordb(sql, tablename, -# "function_input_embedding", execute_sql_vectordb) -# # println(df[1, [:id, :function_output]]) -# row, col = size(df) -# distance = row == 0 ? Inf : df[1, :distance] -# if row != 0 && distance < maxdistance -# # if there is usable SQL, return it. -# output_b64 = df[1, :function_output_base64] # pick the closest match -# output_str = String(base64decode(output_b64)) -# rowid = df[1, :id] -# println("\n--| similar sql found. row id $rowid, distance $distance ", @__FILE__, ":", @__LINE__, " $(Dates.now())") -# pprintln(output_str) -# return (result=output_str, distance=distance) -# else -# println("\n--| similar sql not found, max distance $maxdistance ", @__FILE__, ":", @__LINE__, " $(Dates.now())") -# return (result=nothing, distance=nothing) -# end -# end - -# """ insert query and sql into vector database -# query = "get all wines from wine table" -# sql = "SELECT * FROM wine;" -# insert_sql_vectordb(query, sql) -# """ -# function insert_sql_vectordb(query::T1, SQL::T2; maxdistance::Number=3 -# ) where {T1<:AbstractString, T2<:AbstractString} - -# tablename = "sqlllm_decision_repository" -# # get embedding of the query -# # query = state[:thoughtHistory][:question] -# df = find_similar_text_from_vectordb(query, tablename, -# "function_input_embedding", execute_sql_vectordb) -# row, col = size(df) -# distance = row == 0 ? Inf : df[1, :distance] -# if row == 0 || distance > maxdistance # no close enough SQL stored in the database -# _query_embedding = get_embedding([query]) -# _query_embedding = GeneralUtils.dictify(_query_embedding) -# # println("\n--- _query_embedding() ", @__FILE__, ":", @__LINE__, " $(Dates.now())") -# # println(_query_embedding) -# # println("---\n") -# query_embedding = _query_embedding["data"][1]["embedding"] -# query = replace(query, "'" => "") -# sql_base64 = base64encode(SQL) -# sql_ = replace(SQL, "'" => "") - -# sql = -# """ -# INSERT INTO $tablename (function_input, function_output, function_output_base64, function_input_embedding) VALUES ('$query', '$sql_', '$sql_base64', '$query_embedding'); -# """ -# # println("\n--| added new decision to vectorDB ", @__FILE__, ":", @__LINE__, " $(Dates.now())") -# # println(sql) -# _ = execute_sql_vectordb(sql) -# end -# end - -# """ execute sql against vectordb -# sql = "SELECT * FROM wine;" -# result = execute_sql_vectordb(sql) -# """ -# function execute_sql_vectordb(sql::T) where {T<:AbstractString} -# host_url, _port = split(config["externalservice"]["sommpanion_vectordb"]["url"], ':') -# port = parse(Int, _port) -# dbname = config["externalservice"]["sommpanion_vectordb"]["dbname"] -# user = config["externalservice"]["sommpanion_vectordb"]["user"] -# password = config["externalservice"]["sommpanion_vectordb"]["password"] -# DBconnection = LibPQ.Connection("host=$host_url port=$port dbname=$dbname user=$user password=$password") -# result = LibPQ.execute(DBconnection, sql) -# close(DBconnection) -# return result -# end - -# """ search similar decision llm made from vectordb -# """ -# function similar_sommelier_decision(recentevents::T1; maxdistance::Integer=3 -# )::Union{AbstractDict, Nothing} where {T1<:AbstractString} - -# tablename = "sommelier_decision_repository" -# # find similar -# df = find_similar_text_from_vectordb(recentevents, tablename, -# "function_input_embedding", execute_sql_vectordb) -# row, col = size(df) -# distance = row == 0 ? Inf : df[1, :distance] -# if row != 0 && distance < maxdistance -# # if there is usable decision, return it. -# rowid = df[1, :id] -# println("\n--| found similar decision. row id $rowid, distance $distance ", @__FILE__, " ", @__LINE__) -# output_b64 = df[1, :function_output_base64] # pick the closest match -# _output_str = String(base64decode(output_b64)) -# output = copy(JSON.read(_output_str)) -# return output -# else -# println("\n--| similar decision not found, max distance $maxdistance ", @__FILE__, " ", @__LINE__) -# return nothing -# end -# end - -# """ search similar text from vectordb -# """ -# function find_similar_text_from_vectordb(text::T1, tablename::T2, embeddingColumnName::T3, -# vectorDB::Function; limit::Integer=1 -# )::DataFrame where {T1<:AbstractString, T2<:AbstractString, T3<:AbstractString} -# # get embedding from LLM service -# _embedding = get_embedding([text]) -# _embedding = _embedding["data"][1]["embedding"] -# _embedding = "$_embedding" - -# embedding = _embedding[4:end] # remove 'Any' from Any[...] - -# # check whether there is close enough vector already store in vectorDB. if no, add, else skip -# sql = """ -# SELECT *, $embeddingColumnName <-> '$embedding' as distance -# FROM $tablename -# ORDER BY distance LIMIT $limit; -# """ -# response = vectorDB(sql) -# df = DataFrame(response) - -# return df -# end - -# """ insert decision llm made to vectordb -# """ -# function insert_sommelier_decision(recentevents::T1, decision::T2; maxdistance::Integer=5 -# ) where {T1<:AbstractString, T2<:AbstractDict} -# tablename = "sommelier_decision_repository" -# # find similar -# df = find_similar_text_from_vectordb(recentevents, tablename, -# "function_input_embedding", execute_sql_vectordb) -# row, col = size(df) -# distance = row == 0 ? Inf : df[1, :distance] -# if row == 0 || distance > maxdistance # no close enough SQL stored in the database -# _embedding = get_embedding([recentevents])[1] -# recentevents_embedding = _embedding["data"][1]["embedding"] -# recentevents = replace(recentevents, "'" => "") -# decision_json = JSON.json(decision) -# decision_base64 = base64encode(decision_json) -# decision = replace(decision_json, "'" => "") - -# sql = -# """ -# INSERT INTO $tablename (function_input, function_output, function_output_base64, function_input_embedding) VALUES ('$recentevents', '$decision', '$decision_base64', '$recentevents_embedding'); -# """ -# println("\n--| added new decision to vectorDB ", @__FILE__, " ", @__LINE__) -# println(sql) -# _ = execute_sql_vectordb(sql) -# else -# println("--| similar decision previously cached, distance $distance ", @__FILE__, " ", @__LINE__) -# end -# end - -# function find_related_tables_for_user_question(question::String; top_row_num::Integer=20) - -# metadata_df = GeneralUtils.extract_column_metadata(pg_conn_str) -# embedding_ready = GeneralUtils.generate_embedding_payloads(metadata_df) - -# # use only text content -# embedding_ready_2 = [i["text_content"] for i in embedding_ready] - -# table_embedding = get_embedding(embedding_ready_2) - -# _user_question_embedding = get_embedding([question]) -# user_question_embedding = Float64.(_user_question_embedding["data"][1]["embedding"]) -# user_question_similarity = [] - -# for i in table_embedding["data"] -# i_data = i["embedding"] -# i_float = Float64.(i_data) -# r = 1 - Distances.cosine_dist(i_float, user_question_embedding) -# push!(user_question_similarity, r) -# end - -# new_df = hcat(metadata_df, DataFrame(user_question_similarity = user_question_similarity)) -# sorted_df = sort(new_df, :user_question_similarity, rev=true) # sort max to min -# _top_20_tables = unique(sorted_df[1:top_row_num, :table_name]) -# top_20_tables = [i for i in _top_20_tables] # convert to Vector{String} -# g, id_to_table, table_to_id = GeneralUtils.harvest_db_undirected_schema_graph(pg_conn_str) -# table_relationship = GeneralUtils.resolve_semantic_cluster(top_20_tables, g, table_to_id, id_to_table) - -# # tables that I should put schema in LLM context -# return table_relationship -# end - - -# function prepareContext(state::agentState)::agentContext - -# #TODO filter tools from state.tools based on user intend in user message and tool description -# filteredTools = state.tools - -# #TODO add filtered tools to the current system prompt / modify systemPrompt here -# preparedSystemPrompt = state.systemPrompt - -# #TODO add system prompt, adjust/modify and inject additional context into messages -# preparedMessages = deepcopy(state.messages) # messages that will be send to LLM - -# agentCtx = agentContext(preparedSystemPrompt, preparedMessages, filteredTools) - -# return agentCtx -# end - -# function formatMsgForLLM(ctx::agentContext)::Dict{String, Any} - -# end - - -config = JSON.parsefile("./appconfig.json") -host_url, _port = split(config["externalservice"]["sommpanion_db"]["url"], ':') -port = parse(Int, _port) -dbname = "winedb" -user = config["externalservice"]["sommpanion_db"]["user"] -password = config["externalservice"]["sommpanion_db"]["password"] -pg_conn_str = "host=$host_url port=$port dbname=$dbname user=$user password=$password" -sessionId = "0" -backend_session_topic = "sommpanion.testsubject" -agent_ch = Channel(8) -agent_conn = NATS.connect(config["nats_server_info"]["url"]) - -sub2 = NATS.subscribe(agent_conn, backend_session_topic) do msg - put!(agent_ch, msg) -end - - - -# model=YiemAgent.llmModel("model_1", "unknown", "unknown", "", false, String[], -# YiemAgent.modelCost(0.0, 0.0, 0.0, 0.0), 0, 0) - - - -#WORKING load tools -text2text_llm = text2textInstructLLM(agent_conn, - config["externalservice"]["servicesloadbalancer"]["nats"], - "sender", - config["externalservice"]["fileserver"]["url"]) - -agent = YiemAgent.yiemAgent( - "/home/ton/docker-apps/sommpanion/agent-backend/tools", - text2text_llm -) - - - - - - - - - - -