update

Reviewed-on: #30

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)

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"
 
 
 

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.