update
This commit is contained in:
@@ -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:
|
||||
<NATS debug message>
|
||||
┌ 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"
|
||||
</NATS debug message>
|
||||
|
||||
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")
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+18
-8
@@ -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
|
||||
|
||||
-327
@@ -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
|
||||
)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user