From a3ab288cfeb302e6f28682c4fcd8341d6fc8c679 Mon Sep 17 00:00:00 2001 From: narawat Date: Wed, 22 Jul 2026 10:18:28 +0700 Subject: [PATCH] update --- src/interface.jl | 30 +-- src/llmfunction.jl | 495 ++++++++++++++++++++++++--------------------- src/type.jl | 8 +- 3 files changed, 275 insertions(+), 258 deletions(-) diff --git a/src/interface.jl b/src/interface.jl index 581a0cd..78f6ad9 100644 --- a/src/interface.jl +++ b/src/interface.jl @@ -4,7 +4,7 @@ export addNewMessage, conversation, decisionMaker, reflector, generatechat, generalconversation, detectWineryName, generateSituationReport using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization, - DataFrames, CSV + DataFrames, Serde using GeneralUtils using ..type, ..util, ..llmfunction @@ -122,22 +122,16 @@ function decisionMaker(a::T; recentevents::Integer=20, maxattempt=3 errornote = "N/A" response = nothing # placeholder for show when error msg show up - for attempt in 1:maxattempt - - - msg = Dict( + msg = Dict( "model" => "gemma-4-E4B-it-UD-Q4_K_XL", "messages" => a.chathistory, "temperature" => 0.7 ) + for attempt in 1:maxattempt response = a.context.text2textInstructLLM(a.id, msg) - - response = GeneralUtils.clean_json_response(response) response = GeneralUtils.remove_french_accents(response) - think, response = GeneralUtils.extractthink(response) - response = String(split(response, ", observation")[1]) # in case LLM generate observation key which it isn't supposed to - response = strip(response) + # think, response = GeneralUtils.extractthink(response) # dollar sign in Julia means string interpolation while occursin('$', response) @@ -145,19 +139,13 @@ function decisionMaker(a::T; recentevents::Integer=20, maxattempt=3 end responsedict = nothing - if occursin(requiredKeys[2], response) - try - _responsedict = JSON.parse(response) - responsedict = GeneralUtils.dictify(_responsedict; keytype=String, sort_order=requiredKeys) - catch - println("\nERROR YiemAgent decisionMaker() failed to parse response: $response ", @__FILE__, ":", @__LINE__, " $(Dates.now())") - continue - end - else - println("\nERROR YiemAgent decisionMaker() $errornote --(not qualify response)-> $responsedict ", @__FILE__, ":", @__LINE__, " $(Dates.now())\n") + try + responsedict = Serde.parse_yaml(response) + catch e + println("\nERROR YiemAgent decisionMaker() Error: $e --(not qualify response)-> $response ", @__FILE__, ":", @__LINE__, " $(Dates.now())\n") continue end - + # check whether all answer's key points are in responsedict println("\n---") println(responsedict) diff --git a/src/llmfunction.jl b/src/llmfunction.jl index 5c6c199..61d699d 100644 --- a/src/llmfunction.jl +++ b/src/llmfunction.jl @@ -5,7 +5,7 @@ export virtualWineUserChatbox, jsoncorrection, search_wine_database!, # recomme extractWineAttributes_2, paraphrase, SQLexecution using HTTP, JSON, URIs, Random, PrettyPrinting, UUIDs, Dates, DataFrames, DataStructures, - Base64, Serde + Base64, Serde, LibPQ using GeneralUtils, SQLLLM using ..type, ..util @@ -683,7 +683,9 @@ function predefined_wine_search_sql(a::T, searchterm::String, related_tables = a.context.find_related_tables_for_user_question(searchterm) table_schema = "" for table in related_tables - _table_schema_str = GeneralUtils.get_db_table_schema_simple(a.context.pg_conn_str, table) + _table_schema_str = get_db_table_schema_simple_with_samples(a.context.pg_conn_str, table) + + # _table_schema_str = GeneralUtils.get_db_table_schema_simple(a.context.pg_conn_str, table) table_schema_str = sprint(show, _table_schema_str) * "\n" table_schema = table_schema * table_schema_str end @@ -719,7 +721,14 @@ function predefined_wine_search_sql(a::T, searchterm::String, for attempt in 1:maxattempt response = a.context.text2textInstructLLM("random_id", msg) - responsedict = Serde.parse_yaml(response) + + responsedict = nothing + try + responsedict = Serde.parse_yaml(response) + catch e + println("\nERROR YiemAgent predefined_wine_search_sql() Error: $e --(not qualify response)-> $response ", @__FILE__, ":", @__LINE__, " $(Dates.now())\n") + continue + end # println("\n ", table_schema) println("\n ", responsedict) @@ -750,24 +759,41 @@ function predefined_wine_search_sql(a::T, searchterm::String, for (table_name, table_info_dict) in responsedict for (column_name, v) in table_info_dict - - # - do_not_resolve_BM25_column = ["tasting_notes", "seo_name", "vintage", "grape", "price"] - if column_name ∉ do_not_resolve_BM25_column + bucket = classify_column(a.context.pg_conn_str, table_name, column_name) + + if bucket == "fuzzy_correction" words_catalog = GeneralUtils.harvest_entity_catalog(a.context.pg_conn_str, table_name, column_name) resolved_word = GeneralUtils.resolve_entity(v["value"], words_catalog; threshold=0.9) table_info_dict[column_name]["value"] = resolved_word - else - delete!(responsedict[table_name], column_name) - if length(responsedict[table_name]) == 0 + end + end + end + + + # filter for column that will be used for hard condition (SQL where clause) + # column with "N/A" operator will be used in vector search + vector_search_words = "" + for (table_name, table_dict) in responsedict + for (column_name, column_dict) in table_dict + if column_dict["operator"] ∉ ["=","<>","!=",">","<",">=","<=","!<","!>","<=>"] + vector_search_words = vector_search_words * column_dict["value"] * ", " + delete!(table_dict, column_name) + + # remove table from responsedict if there is no column to used + if length(responsedict[table_name]) == 0 delete!(responsedict, table_name) end end end end - println("\n", responsedict) - @info "after BM25 " @__LINE__ + println("") + pprintln(responsedict) + @info "predefined_wine_search_sql() " @__LINE__ + + #WORKING do vector searched + println("") + @show vector_search_words sql = predefined_wine_search_sql(responsedict) @@ -1260,216 +1286,6 @@ function extractWineAttributes_2(a::T1, input::T2)::String where {T1<:agent, T2< error("extractWineAttributes_2() failed to get a response") end - -function paraphrase(text2textInstructLLM::Function, text::String) - systemmsg = - """ - Your name: N/A - Your vision: - - You are a helpful assistant who help the user to paraphrase their text. - Your mission: - - To help paraphrase the user's text - Mission's objective includes: - - To help paraphrase the user's text - Your responsibility includes: - 1) To help paraphrase the user's text - Your responsibility does NOT includes: - 1) N/A - Your profile: - - N/A - Additional information: - - N/A - - At each round of conversation, you will be given the following information: - Text: The user's given text - - You MUST follow the following guidelines: - - N/A - - You should follow the following guidelines: - - N/A - - You should then respond to the user with: - Paraphrase: Paraphrased text - - You should only respond in format as described below: - Paraphrase: ... - - Let's begin! - """ - #[PENDING] use JSON the same as extractWineAttributes_1 is better. change this function to use the same format use decisionMaker - header = ["Paraphrase:"] - dictkey = ["paraphrase"] - - errornote = "N/A" - response = nothing # placeholder for show when error msg show up - - - for attempt in 1:10 - usermsg = """ - Text: $text - P.S. $errornote - """ - - _prompt = - [ - Dict("name" => "system", "text" => systemmsg), - Dict("name" => "user", "text" => usermsg) - ] - - # put in model format - prompt = GeneralUtils.formatLLMtext(_prompt, a.llmFormatName) - - try - response = text2textInstructLLM(prompt) - response = GeneralUtils.deFormatLLMtext(response, a.llmFormatName) - think, response = GeneralUtils.extractthink(response) - # sometime the model response like this "here's how I would respond: ..." - if occursin("respond:", response) - errornote = "You don't need to intro your response" - error("\nparaphrase() response contain : ", @__FILE__, ":", @__LINE__, " $(Dates.now())") - end - response = GeneralUtils.remove_french_accents(response) - response = replace(response, '*'=>"") - response = replace(response, '$' => "USD") - response = replace(response, '`' => "") - response = GeneralUtils.remove_french_accents(response) - - # check whether response has all answer's key points - detected_kw = GeneralUtils.detect_keyword(header, response) - if 0 ∈ values(detected_kw) - errornote = "\nYiemAgent paraphrase() response does not have all answer's key points" - continue - elseif sum(values(detected_kw)) > length(header) - errornote = "\nnYiemAgent paraphrase() response has duplicated answer's key points" - continue - end - - responsedict = GeneralUtils.textToDict(response, header; - dictKey=dictkey, symbolkey=true) - - for i ∈ [:paraphrase] - if length(JSON.json(responsedict[i])) == 0 - error("$i is empty ", @__FILE__, ":", @__LINE__, " $(Dates.now())") - end - end - - # check if there are more than 1 key per categories - for i ∈ [:paraphrase] - matchkeys = GeneralUtils.findMatchingDictKey(responsedict, i) - if length(matchkeys) > 1 - error("paraphrase() has more than one key per categories") - end - end - - println("\nparaphrase() ", @__FILE__, ":", @__LINE__, " $(Dates.now())") - pprintln(Dict(responsedict)) - - result = responsedict["paraphrase"] - - return result - catch e - io = IOBuffer() - showerror(io, e) - errorMsg = String(take!(io)) - st = sprint((io, v) -> show(io, "text/plain", v), stacktrace(catch_backtrace())) - println("\nAttempt $attempt. Error occurred: $errorMsg\n$st ", @__FILE__, ":", @__LINE__, " $(Dates.now())") - end - end - error("paraphrase() failed to generate a response") -end - - - -""" Attemp to correct LLM response's incorrect JSON response. - -# Arguments - - `a::T1` - one of Yiem's agent - - `input::T2` - text to be send to virtual wine customer - -# Return - - `correctjson::String` - corrected json string - -# Example -```jldoctest -julia> -``` - -# Signature -""" -function jsoncorrection(config::T1, input::T2, correctJsonExample::T3; - maxattempt::Integer=3 - ) where {T1<:AbstractDict, T2<:AbstractString, T3<:AbstractString} - - incorrectjson = deepcopy(input) - correctjson = nothing - - for attempt in 1:maxattempt - try - d = copy(JSON.parsefile(incorrectjson)) - correctjson = incorrectjson - return correctjson - catch e - @warn "Attempting to correct JSON string. Attempt $attempt" - e = """$e""" - if occursin("EOF", e) - e = split(e, "EOF")[1] * "EOF" - end - incorrectjson = deepcopy(input) - _prompt = - """ - Your goal are: - 1) Use the expected JSON format as a guideline to check why the given JSON string failed to load and provide a corrected version that can be loaded by Python's json.load function. - 2) Provide Corrected JSON string only. Do not provide any other info. - - $correctJsonExample - - Let's begin! - Given JSON string: $incorrectjson - The given JSON string failed to load previously because: $e - Corrected JSON string: - """ - - # apply LLM specific instruct format - externalService = config["externalservice"]["text2textinstruct"] - llminfo = externalService["llminfo"] - prompt = - if llminfo["name"] == "llama3instruct" - formatLLMtext_llama3instruct("system", _prompt) - else - error("llm model name is not defied yet $(@__LINE__)") - end - - # send formatted input to user using GeneralUtils.sendReceiveMqttMsg - msgMeta = GeneralUtils.generate_msgMeta( - externalService["mqtttopic"], - senderName= "jsoncorrection", - senderId= string(uuid4()), - receiverName= "text2textinstruct", - mqttBroker= config["mqttServerInfo"]["broker"], - mqttBrokerPort= config["mqttServerInfo"]["port"], - ) - - outgoingMsg = Dict( - "msgMeta"=> msgMeta, - "payload"=> Dict( - "text"=> prompt, - "kwargs"=> Dict( - "max_tokens"=> 512, - "stop"=> ["<|eot_id|>"], - ) - ) - ) - result = GeneralUtils.sendReceiveMqttMsg(outgoingMsg; timeout=120) - incorrectjson = result[:response][:text] - end - end -end - - function predefined_wine_search_sql(conditions::Dict{String, Any})::String # 1. Base SQL structure base_query = @@ -1492,11 +1308,11 @@ SELECT rw.price, rw.currency, w.image_url, - NULL AS retailer_name,66 + r.retailer_name, rw.retailer_id FROM wine AS w -JOIN retailer_wine AS rw - ON w.wine_id = rw.wine_id +JOIN retailer_wine AS rw ON w.wine_id = rw.wine_id +JOIN retailer AS r ON rw.retailer_id = r.retailer_id """ # 2. Dynamic WHERE Clause Builder @@ -1560,6 +1376,94 @@ end +function get_db_table_schema_simple_with_samples(pg_conn_str::String, table_name::String; + schema_name::String="public")::String + conn = LibPQ.Connection(pg_conn_str) + return get_db_table_schema_simple_with_samples(conn, table_name; schema_name=schema_name) +end + +function get_db_table_schema_simple_with_samples(conn, table_name::String; schema_name::String="public", sample_count::Int=3)::String + # 1. SQL query for catalog metadata + meta_sql = """ + SELECT + a.attname AS column_name, + format_type(a.atttypid, a.atttypmod) AS data_type, + pg_get_expr(def.adbin, def.adrelid) AS default_value, + COALESCE( + (SELECT pg_get_constraintdef(p.oid) + FROM pg_catalog.pg_constraint p + WHERE p.conrelid = c.oid AND a.attnum = ANY(p.conkey) + LIMIT 1), '' + ) AS constraint_definition + FROM pg_catalog.pg_attribute a + JOIN pg_catalog.pg_class c ON a.attrelid = c.oid + JOIN pg_catalog.pg_namespace n ON c.relnamespace = n.oid + LEFT JOIN pg_catalog.pg_attrdef def ON def.adrelid = c.oid AND def.adnum = a.attnum + WHERE c.relname = \$1 + AND n.nspname = \$2 + AND a.attnum > 0 + AND NOT a.attisdropped + ORDER BY a.attnum; + """ + + meta_res = DataFrame(execute(conn, meta_sql, [table_name, schema_name])) + + if nrow(meta_res) == 0 + error("Table '$schema_name.$table_name' not found.") + end + + # 2. Build single dynamic query to fetch non-null samples for all columns + sample_selects = String[] + for row in eachrow(meta_res) + c_name = row.column_name + push!(sample_selects, """ + (SELECT json_agg(s."$c_name") + FROM ( + SELECT "$c_name" + FROM "$schema_name"."$table_name" + WHERE "$c_name" IS NOT NULL + LIMIT $sample_count + ) s + ) AS "$c_name" + """) + end + + sample_sql = "SELECT " * join(sample_selects, ",\n ") * ";" + sample_df = DataFrame(execute(conn, sample_sql)) + + # 3. Build DDL definitions with inline sample comments + ddl_lines = String[] + constraints = String[] + + for row in eachrow(meta_res) + col_name = row.column_name + data_type = row.data_type + default_val = ismissing(row.default_value) ? "" : " DEFAULT " * row.default_value + + col_def = " \"$col_name\" $data_type$default_val" + + # Fetch sample data for this column from the single-row sample DataFrame + samples_comment = "" + if nrow(sample_df) > 0 + raw_samples = sample_df[1, Symbol(col_name)] + samples_str = ismissing(raw_samples) || isnothing(raw_samples) ? "[]" : string(raw_samples) + samples_comment = " -- Samples: $samples_str" + end + + push!(ddl_lines, col_def * samples_comment) + + # Handle table-level constraints + con_def = ismissing(row.constraint_definition) ? "" : row.constraint_definition + if !isempty(con_def) && !(con_def in constraints) + push!(constraints, " " * con_def) + end + end + + all_definitions = vcat(ddl_lines, constraints) + body = join(all_definitions, ",\n") + + return "CREATE TABLE \"$schema_name\".\"$table_name\" (\n$body\n);" +end @@ -1570,20 +1474,145 @@ end +function classify_column(pg_conn_str::String, table_name::String, column_name::String; + sample_size::Integer=1000) + conn = LibPQ.Connection(pg_conn_str) + return classify_column(conn, table_name, column_name; sample_size=sample_size) +end + + +function classify_column(conn::LibPQ.Connection, table_name::String, column_name::String; sample_size::Int=1000) + # 1. Fetch BOTH data_type and udt_name (User Defined Type name) + meta_query = """ + SELECT data_type, udt_name + FROM information_schema.columns + WHERE table_name = lower('$(table_name)') + AND column_name = lower('$(column_name)'); + """ + + pg_type = "unknown" + udt_name = "unknown" + + try + df = DataFrame(LibPQ.execute(conn, meta_query)) + if !isempty(df) + pg_type = df[1, :data_type] + udt_name = df[1, :udt_name] + end + catch e + @error "Failed to fetch metadata for $table_name.$column_name" exception=e + return "error" + end + + # 2. FAST-TRACK: Check for pgvector FIRST + # pgvector registers as "USER-DEFINED" in data_type, but "vector" in udt_name + if udt_name == "vector" + return "semantic_search" + end + + # 3. FAST-TRACK: Hard rules for standard non-text Postgres types + if pg_type in ["integer", "bigint", "smallint", "numeric", "real", + "double precision", "boolean", "date", + "timestamp without time zone", "timestamp with time zone", "uuid"] + return "exact_or_range" + end + + # 4. SAMPLE: Get text statistics for remaining text columns + stats_query = """ + SELECT + COUNT(*)::int AS total_count, + COUNT(DISTINCT $(column_name)::text)::int AS unique_count, + COALESCE(AVG(LENGTH($(column_name)::text)), 0)::float AS avg_len, + COALESCE(STDDEV(LENGTH($(column_name)::text)), 0)::float AS std_len + FROM ( + SELECT $(column_name) + FROM $(table_name) + WHERE $(column_name) IS NOT NULL + LIMIT $sample_size + ) AS sampled_data; + """ + + try + df = DataFrame(LibPQ.execute(conn, stats_query)) + if isempty(df) || df[1, :total_count] == 0 + return "unknown" + end + + total = df[1, :total_count] + unique = df[1, :unique_count] + avg_len = df[1, :avg_len] + std_len = df[1, :std_len] + ratio = unique / total + + # 5. HEURISTICS: Route the column_name to the correct text bucket + return classify_text_column(unique, ratio, avg_len, std_len) + + catch e + @warn "Failed to sample column_name $table_name.$column_name" exception=e + return "unknown" + end +end + +# The Decision Tree for Text Columns (Unchanged, but kept for completeness) +function classify_text_column(unique_count::Integer, ratio::Float64, avg_len::Float64, std_len::Float64) + if avg_len > 60 && std_len > 25 + return "full_text_search" + end + if ratio > 0.90 && avg_len < 40 + return "exact_or_regex" + end + if unique_count <= 100 + return "fuzzy_correction" + end + if ratio > 0.10 && avg_len < 35 + return "fuzzy_correction" + end + if avg_len < 60 + return "fuzzy_correction" + end + return "full_text_search" +end +function harvest_entity_catalog(pg_conn_str::String, table::String, column::String) + conn = LibPQ.Connection(pg_conn_str) + return harvest_entity_catalog(conn, table, column) +end +function harvest_entity_catalog_with_pg_type(conn::LibPQ.Connection, table::String, column::String) + try + # 1. Query the actual data + data_query = "SELECT DISTINCT $(column) FROM $(table) WHERE $(column) IS NOT NULL;" + df = DataFrame(LibPQ.execute(conn, data_query)) + values = String.(strip.(string.(df[!, 1]))) + + # 2. Query the database schema for the column's data type + # Note: Postgres stores unquoted table/column names in lowercase + type_query = """ + SELECT data_type + FROM information_schema.columns + WHERE table_name = lower('$(table)') + AND column_name = lower('$(column)'); + """ + type_df = DataFrame(LibPQ.execute(conn, type_query)) + pg_type = isempty(type_df) ? "unknown" : type_df[1, 1] + + return (values = values, type = pg_type) + + catch e + @error "Failed to harvest catalog" exception=e + return (values = String[], type = "unknown") + finally + close(conn) + end +end - - - - - - - - +# Usage: +# result = harvest_entity_catalog_with_pg_type(conn, "users", "created_at") +# println(result.values) # ["2023-01-01", "2023-02-15"] +# println(result.type) # "timestamp without time zone" diff --git a/src/type.jl b/src/type.jl index a906fb6..452ecb6 100644 --- a/src/type.jl +++ b/src/type.jl @@ -206,10 +206,10 @@ function sommelier( 3) "action_input", The input to the action you are about to perform according to your plan. After the action is executed you gets "action_result". It is the output from the action you selected. - # you should only respond in JSON format as described below (not Markdown format) - "plan": "...", - "action_name": "...", - "action_input": "..." + # you should only respond in YAML format as described below + plan: "..." + action_name: "..." + action_input: "..." # available actions "CHAT_BOX", which you can use to talk with the user. The input is dialogue you want to chat with the user according to your plan.