Hologres：最佳實務：金融多模態AI資料分析與檢索系統

PDF加工預存程序

• 建立Object Table和Dynamic Table

  CREATE OR REPLACE PROCEDURE create_rag_corpus_from_oss(
      oss_path TEXT,
      oss_endpoint TEXT,
      oss_role_arn TEXT,
      corpus_table TEXT,
      embedding_model TEXT DEFAULT NULL,
      parse_document_model TEXT DEFAULT NULL,
      chunk_model TEXT DEFAULT NULL,
      chunk_size INT DEFAULT 300,
      chunk_overlap INT DEFAULT 50,
      overwrite BOOLEAN DEFAULT FALSE
  )
  AS $$
  DECLARE
      corpus_schema TEXT;
      corpus_name TEXT;
      obj_table_name TEXT;
      full_corpus_ident TEXT;
      full_obj_ident TEXT;
      embed_expr TEXT;
      chunk_expr TEXT;
      parse_expr TEXT;
      embedding_dims INT;
  BEGIN
      -- 1. 拆 schema name + table name
      IF position('.' in corpus_table) > 0 THEN
          corpus_schema := split_part(corpus_table, '.', 1);
          corpus_name   := split_part(corpus_table, '.', 2);
      ELSE
          corpus_schema := 'public';
          corpus_name   := corpus_table;
      END IF;
  
      obj_table_name := corpus_name || '_obj_table';
  
      full_corpus_ident := format('%I.%I', corpus_schema, corpus_name);
      full_obj_ident    := format('%I.%I', corpus_schema, obj_table_name);
      
      -- 2. 如果需要覆蓋，先刪表和索引
      IF overwrite THEN
          DECLARE
              dyn_table_exists BOOLEAN;
              rec RECORD;
          BEGIN
              -- 檢查 dynamic table 是否存在
              SELECT EXISTS (
                  SELECT 1
                  FROM pg_class c
                  JOIN pg_namespace n ON n.oid = c.relnamespace
                  WHERE c.relname = corpus_name
                  AND n.nspname = corpus_schema
              )
              INTO dyn_table_exists;
  
              IF dyn_table_exists THEN
                  -- 2.1 關閉動態表自動重新整理
                  -- RAISE NOTICE 'Disabling auto refresh for %', full_corpus_ident;
                  -- EXECUTE format('ALTER TABLE IF EXISTS %s SET (auto_refresh_enable=false)', full_corpus_ident);
  
                  -- 2.2 尋找 RUNNING 重新整理任務並取消
                  FOR rec IN
                      EXECUTE format(
                          $f$
                          SELECT query_job_id
                              FROM hologres.hg_dynamic_table_refresh_log(%L)
                              WHERE status = 'RUNNING';
                          $f$,
                          corpus_table
                      )
                  LOOP
                      RAISE NOTICE 'Found running refresh job: %', rec.query_job_id;
                      IF hologres.hg_internal_cancel_query_job(rec.query_job_id::bigint) THEN
                          RAISE NOTICE 'Cancel job % succeeded.', rec.query_job_id;
                      ELSE
                          RAISE WARNING 'Cancel job % failed.', rec.query_job_id;
                      END IF;
                  END LOOP;
  
                  -- 2.3 刪除 Dynamic Table
                  EXECUTE format('DROP TABLE IF EXISTS %s;', full_corpus_ident);
              ELSE
                  RAISE NOTICE 'Dynamic table % does not exist, skip cancel job and drop.', full_corpus_ident;
              END IF;
  
              -- 2.4 無論如何，Object Table 都要刪除
              EXECUTE format('DROP OBJECT TABLE IF EXISTS %s;', full_obj_ident);
          END;
      END IF;
  
      -- 3. 建立 Object Table
      RAISE NOTICE 'Create object table: %', obj_table_name;
      EXECUTE format(
          $f$
          CREATE OBJECT TABLE %s
          WITH (
              path = %L,
              oss_endpoint = %L,
              role_arn = %L
          );
          $f$,
          full_obj_ident,
          oss_path,
          oss_endpoint,
          oss_role_arn
      );
  
      COMMIT;
  
      -- 4. 重新整理 Object Table
      RAISE NOTICE 'Refresh object table: %', obj_table_name;
      EXECUTE format('REFRESH OBJECT TABLE %s;', full_obj_ident);
  
      COMMIT;
  
      -- 5. 文檔解析模型選擇
      IF parse_document_model IS NULL OR length(trim(parse_document_model)) = 0 THEN
          parse_expr := 'ai_parse_document(file, ''auto'', ''markdown'')';
      ELSE
          parse_expr := format(
              'ai_parse_document(%L, file, ''auto'', ''markdown'')',
              parse_document_model
          );
      END IF;
  
      -- 6. chunk 模型選擇
      IF chunk_model IS NULL OR length(trim(chunk_model)) = 0 THEN
          chunk_expr := format('ai_chunk(doc, %s, %s)', chunk_size, chunk_overlap);
      ELSE
          chunk_expr := format(
              'ai_chunk(%L, doc, %s, %s)',
              chunk_model,
              chunk_size,
              chunk_overlap
          );
      END IF;
  
      -- 7. embedding 模型選擇
      IF embedding_model IS NULL OR length(trim(embedding_model)) = 0 THEN
          embed_expr := 'ai_embed(chunk)';
  
          EXECUTE 'SELECT array_length(ai_embed(''dummy''), 1)'
          INTO embedding_dims;
      ELSE
          embed_expr := format('ai_embed(%L, chunk)', embedding_model);
  
          EXECUTE format(
              'SELECT array_length(ai_embed(%L, ''dummy''), 1)',
              embedding_model
          )
          INTO embedding_dims;
      END IF;
  
      RAISE NOTICE 'embedding dimension is: %', embedding_dims;
  
      -- 8. 建立 RAG 輸出動態表
      RAISE NOTICE 'create dynamic table: %', corpus_name;
      EXECUTE format(
          $f$
          CREATE DYNAMIC TABLE %s(
              CHECK(array_ndims(embedding_vector) = 1 AND array_length(embedding_vector, 1) = %s)
          )
          WITH (
              vectors = '{
                "embedding_vector": {
                  "algorithm": "HGraph",
                  "distance_method": "Cosine",
                  "builder_params": {
                  "base_quantization_type": "sq8_uniform",
                  "max_degree": 64,
                  "ef_construction": 400,
                  "precise_quantization_type": "fp32",
                  "use_reorder": true
                  }
                }
              }',
              auto_refresh_mode = 'incremental',
              freshness = '5 minutes',
              auto_refresh_enable = 'false'
          ) AS
          WITH parsed_doc AS (
              SELECT object_uri,
                     etag,
                     %s AS doc
                FROM %s
          ),
          chunked_doc AS (
              SELECT object_uri,
                     etag,
                     unnest(%s) AS chunk
                FROM parsed_doc
          )
          SELECT
              object_uri,
              etag,
              chunk,
              %s AS embedding_vector
            FROM chunked_doc;
          $f$,
          full_corpus_ident,
          embedding_dims,
          parse_expr,
          full_obj_ident,
          chunk_expr,
          embed_expr
      );
      COMMIT;
  
      -- 9. 建立全文索引（索引名 = 表名 || '_fulltext_idx'）
      EXECUTE format(
          'CREATE INDEX %I ON %s USING FULLTEXT (chunk);',
          corpus_name || '_fulltext_idx',
          full_corpus_ident
      );
  
      RAISE NOTICE '';
      RAISE NOTICE 'Create RAG corpus success to table: %', corpus_table;
      RAISE NOTICE '    Vector index is: %.embedding_vector', corpus_table;
      RAISE NOTICE '    TextSearch index is: %.chunk', corpus_table;
  END;
  $$ LANGUAGE plpgsql;

• 重新整理Object Table和Dynamic Table預存程序

  CREATE OR REPLACE PROCEDURE refresh_rag_corpus_table(
      corpus_table TEXT
  )
  AS $$
  DECLARE
      corpus_schema TEXT;
      corpus_name   TEXT;
      obj_table_name TEXT;
      full_corpus_ident TEXT;
      full_obj_ident    TEXT;
  BEGIN
      -- 1. 解析 schema 和表名
      IF position('.' in corpus_table) > 0 THEN
          corpus_schema := split_part(corpus_table, '.', 1);
          corpus_name   := split_part(corpus_table, '.', 2);
      ELSE
          corpus_schema := 'public';
          corpus_name   := corpus_table;
      END IF;
  
      obj_table_name := corpus_name || '_obj_table';
  
      full_corpus_ident := format('%I.%I', corpus_schema, corpus_name);
      full_obj_ident    := format('%I.%I', corpus_schema, obj_table_name);
  
      -- 2. 重新整理 Object Table
      RAISE NOTICE 'Refreshing Object Table: %', obj_table_name;
      EXECUTE format('REFRESH OBJECT TABLE %s;', full_obj_ident);
  
      -- 3. 重新整理 Dynamic Table
      RAISE NOTICE 'Refreshing Dynamic Table: %', corpus_name;
      EXECUTE format('REFRESH TABLE %s;', full_corpus_ident);
  
      RAISE NOTICE 'Refresh complete for corpus table %', corpus_table;
  END;
  $$ LANGUAGE plpgsql;

• 刪除Object Table和Dynamic Table預存程序

  CREATE OR REPLACE PROCEDURE drop_rag_corpus_table(
      corpus_table TEXT
  )
  AS $$
  DECLARE
      corpus_schema TEXT;
      corpus_name   TEXT;
      obj_table_name TEXT;
      full_corpus_ident TEXT;
      full_obj_ident    TEXT;
      rec RECORD;
  BEGIN
      -- 1. 解析 schema 和表名
      IF position('.' in corpus_table) > 0 THEN
          corpus_schema := split_part(corpus_table, '.', 1);
          corpus_name   := split_part(corpus_table, '.', 2);
      ELSE
          corpus_schema := 'public';
          corpus_name   := corpus_table;
      END IF;
  
      obj_table_name := corpus_name || '_obj_table';
  
      full_corpus_ident := format('%I.%I', corpus_schema, corpus_name);
      full_obj_ident    := format('%I.%I', corpus_schema, obj_table_name);
  
      -- 2. 刪除表
      -- 2.1 關閉動態表自動重新整理
      -- RAISE NOTICE 'Disabling auto refresh for %', full_corpus_ident;
      -- EXECUTE format('ALTER TABLE IF EXISTS %s SET (auto_refresh_enable=false)', full_corpus_ident);
  
      -- 2.2 尋找 RUNNING 重新整理任務並取消
      FOR rec IN
          EXECUTE format(
              $f$
              SELECT query_job_id
                  FROM hologres.hg_dynamic_table_refresh_log(%L)
                  WHERE status = 'RUNNING';
              $f$,
              corpus_table
          )
      LOOP
          RAISE NOTICE 'Found running refresh job: %', rec.query_job_id;
          IF hologres.hg_internal_cancel_query_job(rec.query_job_id::bigint) THEN
              RAISE NOTICE 'Cancel job % succeeded.', rec.query_job_id;
          ELSE
              RAISE WARNING 'Cancel job % failed.', rec.query_job_id;
          END IF;
      END LOOP;
  
      -- 2.3 刪除 Dynamic Table
      RAISE NOTICE 'Dropping Dynamic Table: %', corpus_name;
      EXECUTE format('DROP TABLE IF EXISTS %s;', full_corpus_ident);
  
      -- 2.4 刪除 Object Table
      RAISE NOTICE 'Dropping Object Table: %', obj_table_name;
      EXECUTE format('DROP OBJECT TABLE IF EXISTS %s;', full_obj_ident);
  
      RAISE NOTICE 'Drop complete for corpus: %', corpus_table;
  END;
  $$ LANGUAGE plpgsql;

向量檢索函數

-- RAG向量單路召回問答
CREATE OR REPLACE FUNCTION qa_vector_search_retrieval(
    question TEXT,
    corpus_table TEXT,
    embedding_model TEXT DEFAULT NULL,
    llm_model TEXT DEFAULT NULL,
    ranking_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT 'Please answer the following question in ${language} based on the reference information.\n\n Question: ${question}\n\n Reference information:\n\n ${context}',
    language TEXT DEFAULT 'Chinese',
    vector_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5,
    vector_col TEXT DEFAULT 'embedding_vector'
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    sql TEXT;
    embedding_expr TEXT;
    ai_rank_expr TEXT;
    ai_gen_expr TEXT;
    embedding_model_valid BOOLEAN;
    llm_model_valid BOOLEAN;
    ranking_model_valid BOOLEAN;
BEGIN
    embedding_model_valid := (embedding_model IS NOT NULL AND trim(embedding_model) != '');
    llm_model_valid := (llm_model IS NOT NULL AND trim(llm_model) != '');
    ranking_model_valid := (ranking_model IS NOT NULL AND trim(ranking_model) != '');

    IF embedding_model_valid THEN
        embedding_expr := 'ai_embed(' || quote_literal(embedding_model) || ', ' || quote_literal(question) || ')';
    ELSE
        embedding_expr := 'ai_embed(' || quote_literal(question) || ')';
    END IF;

    IF ranking_model_valid THEN
        ai_rank_expr := 'ai_rank(' || quote_literal(ranking_model) || ', ' || quote_literal(question) || ', chunk)';
    ELSE
        ai_rank_expr := 'ai_rank(' || quote_literal(question) || ', chunk)';
    END IF;

    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(replace(' || quote_literal(prompt) ||
               ', ''${question}'', ' || quote_literal(question) || '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || ') )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(replace(' || quote_literal(prompt) ||
               ', ''${question}'', ' || quote_literal(question) || '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || '))';
    END IF;

    sql := '
      WITH
        embedding_recall AS (
            SELECT
              chunk,
              approx_cosine_distance(' || vector_col || ', ' || embedding_expr || ') AS distance
            FROM
              ' || corpus_table || '
            ORDER BY
              distance DESC
            LIMIT ' || vector_recall_count || '
        ),
        rerank AS (
            SELECT
              chunk,
              ' || ai_rank_expr || ' AS score
            FROM
              embedding_recall
            ORDER BY
              score DESC
            LIMIT ' || rerank_recall_count || '
        ),
        concat_top_chunks AS (
            SELECT string_agg(chunk, E''\n\n----\n\n'') AS merged_chunks FROM rerank
        )
      SELECT ' || ai_gen_expr || '
      FROM concat_top_chunks;
    ';

    EXECUTE sql INTO final_answer;
    RETURN final_answer;
END;
$$ LANGUAGE plpgsql;

全文檢索索引函數

CREATE OR REPLACE FUNCTION qa_text_search_retrieval(
    question TEXT,
    corpus_table TEXT,
    llm_model TEXT DEFAULT NULL,
    ranking_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT 'Please answer the following question in ${language} based on the reference information.\n\n Question: ${question}\n\n Reference information:\n\n ${context}',
    language TEXT DEFAULT 'Chinese',
    text_search_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5,
    text_search_col TEXT DEFAULT 'chunk'
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    sql TEXT;
    ai_rank_expr TEXT;
    ai_gen_expr TEXT;
    llm_model_valid BOOLEAN;
    ranking_model_valid BOOLEAN;
BEGIN
    llm_model_valid     := (llm_model IS NOT NULL AND trim(llm_model) != '');
    ranking_model_valid := (ranking_model IS NOT NULL AND trim(ranking_model) != '');

    IF ranking_model_valid THEN
        ai_rank_expr := 'ai_rank(' || quote_literal(ranking_model) || ', ' || quote_literal(question) || ', chunk)';
    ELSE
        ai_rank_expr := 'ai_rank(' || quote_literal(question) || ', chunk)';
    END IF;

    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(replace(' || quote_literal(prompt) ||
               ', ''${question}'', ' || quote_literal(question) ||
               '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || ') )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(replace(' || quote_literal(prompt) ||
               ', ''${question}'', ' || quote_literal(question) ||
               '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || '))';
    END IF;

    sql := '
      WITH
        text_search_recall AS (
            SELECT
              chunk
            FROM
              ' || corpus_table || '
            ORDER BY
              text_search(' || text_search_col || ', ' || quote_literal(question) || ') DESC
            LIMIT ' || text_search_recall_count || '
        ),
        rerank AS (
            SELECT
              chunk,
              ' || ai_rank_expr || ' AS score
            FROM
              text_search_recall
            ORDER BY
              score DESC
            LIMIT ' || rerank_recall_count || '
        ),
        concat_top_chunks AS (
            SELECT string_agg(chunk, E''\n\n----\n\n'') AS merged_chunks FROM rerank
        )
      SELECT ' || ai_gen_expr || '
      FROM concat_top_chunks;
    ';

    EXECUTE sql INTO final_answer;
    RETURN final_answer;
END;
$$ LANGUAGE plpgsql;

向量+全文雙路召回+Rank排序函數

CREATE OR REPLACE FUNCTION qa_hybrid_retrieval(
    question TEXT,
    corpus_table TEXT,
    embedding_model TEXT DEFAULT NULL,
    llm_model TEXT DEFAULT NULL,
    ranking_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT 'Please answer the following question in ${language} based on the reference information.\n\n Question: ${question}\n\n Reference information:\n\n ${context}',
    language TEXT DEFAULT 'Chinese',
    text_search_recall_count INT DEFAULT 20,
    vector_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5,
    vector_col TEXT DEFAULT 'embedding_vector',
    text_search_col TEXT DEFAULT 'chunk'
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    sql TEXT;
    embedding_expr TEXT;
    ai_rank_expr TEXT;
    ai_gen_expr TEXT;
    embedding_model_valid BOOLEAN;
    llm_model_valid BOOLEAN;
    ranking_model_valid BOOLEAN;
BEGIN
    embedding_model_valid := (embedding_model IS NOT NULL AND trim(embedding_model) != '');
    llm_model_valid       := (llm_model IS NOT NULL AND trim(llm_model) != '');
    ranking_model_valid   := (ranking_model IS NOT NULL AND trim(ranking_model) != '');

    IF embedding_model_valid THEN
        embedding_expr := 'ai_embed(' || quote_literal(embedding_model) || ', ' || quote_literal(question) || ')';
    ELSE
        embedding_expr := 'ai_embed(' || quote_literal(question) || ')';
    END IF;

    IF ranking_model_valid THEN
        ai_rank_expr := 'ai_rank(' || quote_literal(ranking_model) || ', ' || quote_literal(question) || ', chunk)';
    ELSE
        ai_rank_expr := 'ai_rank(' || quote_literal(question) || ', chunk)';
    END IF;

    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(replace(' || quote_literal(prompt) ||
               ', ''${question}'', ' || quote_literal(question) || '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || ') )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(replace(' || quote_literal(prompt) ||
               ', ''${question}'', ' || quote_literal(question) || '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || '))';
    END IF;

    sql := '
      WITH
        embedding_recall AS (
            SELECT
              chunk
            FROM
              ' || corpus_table || '
            ORDER BY
              approx_cosine_distance(' || vector_col || ', ' || embedding_expr || ') DESC
            LIMIT ' || vector_recall_count || '
        ),
        text_search_recall AS (
            SELECT
              chunk
            FROM
              ' || corpus_table || '
            ORDER BY
              text_search(' || text_search_col || ', ' || quote_literal(question) || ') DESC
            LIMIT ' || text_search_recall_count || '
        ),
        union_recall AS (
            SELECT chunk FROM embedding_recall
            UNION
            SELECT chunk FROM text_search_recall
        ),
        rerank AS (
            SELECT
              chunk,
              ' || ai_rank_expr || ' AS score
            FROM
              union_recall
            ORDER BY
              score DESC
            LIMIT ' || rerank_recall_count || '
        ),
        concat_top_chunks AS (
            SELECT string_agg(chunk, E''\n\n----\n\n'') AS merged_chunks FROM rerank
        )
      SELECT ' || ai_gen_expr || '
      FROM concat_top_chunks;
    ';

    EXECUTE sql INTO final_answer;
    RETURN final_answer;
END;
$$ LANGUAGE plpgsql;

向量+全文雙路召回+RRF排序函數

CREATE OR REPLACE FUNCTION qa_hybrid_retrieval_rrf(
    question TEXT,
    corpus_table TEXT,
    embedding_model TEXT DEFAULT NULL,
    llm_model TEXT DEFAULT NULL,
    ranking_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT 'Please answer the following question in ${language} based on the reference information.\n\n Question: ${question}\n\n Reference information:\n\n ${context}',
    language TEXT DEFAULT 'Chinese',
    text_search_recall_count INT DEFAULT 20,
    vector_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5,
    rrf_k INT DEFAULT 60,
    vector_col TEXT DEFAULT 'embedding_vector',
    text_search_col TEXT DEFAULT 'chunk'
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    sql TEXT;
    embedding_expr TEXT;
    ai_rank_expr TEXT;
    ai_gen_expr TEXT;
    embedding_model_valid BOOLEAN;
    llm_model_valid BOOLEAN;
    ranking_model_valid BOOLEAN;
BEGIN
    embedding_model_valid := (embedding_model IS NOT NULL AND trim(embedding_model) <> '');
    llm_model_valid       := (llm_model IS NOT NULL AND trim(llm_model) <> '');
    ranking_model_valid   := (ranking_model IS NOT NULL AND trim(ranking_model) <> '');

    IF embedding_model_valid THEN
        embedding_expr := 'ai_embed(' || quote_literal(embedding_model) || ', ' || quote_literal(question) || ')';
    ELSE
        embedding_expr := 'ai_embed(' || quote_literal(question) || ')';
    END IF;

    IF ranking_model_valid THEN
        ai_rank_expr := 'ai_rank(' || quote_literal(ranking_model) || ', ' || quote_literal(question) || ', chunk)';
    ELSE
        ai_rank_expr := 'ai_rank(' || quote_literal(question) || ', chunk)';
    END IF;

    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(replace(' || quote_literal(prompt) ||
                ', ''${question}'', ' || quote_literal(question) || '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || ') )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(replace(' || quote_literal(prompt) ||
                ', ''${question}'', ' || quote_literal(question) || '), ''${context}'', merged_chunks), ''${language}'', ' || quote_literal(language) || '))';
    END IF;

    sql := '
      WITH embedding_recall AS (
        SELECT
          chunk,
          vec_score,
          ROW_NUMBER() OVER (ORDER BY vec_score DESC) AS rank_vec
        FROM (
          SELECT
            chunk,
            approx_cosine_distance(' || vector_col || ', ' || embedding_expr || ') AS vec_score
          FROM
            ' || corpus_table || '
        ) t
        ORDER BY vec_score DESC
        LIMIT ' || vector_recall_count || '
      ),
      text_search_recall AS (
        SELECT
          chunk,
          text_score,
          ROW_NUMBER() OVER (ORDER BY text_score DESC) AS rank_text
        FROM (
          SELECT
            chunk,
            text_search(' || text_search_col || ', ' || quote_literal(question) || ') AS text_score
          FROM
            ' || corpus_table || '
        ) ts
        WHERE text_score > 0
        ORDER BY text_score DESC
        LIMIT ' || text_search_recall_count || '
      ),
      rrf_scores AS (
        SELECT
          chunk,
          SUM(1.0 / (' || rrf_k || ' + rank_val)) AS rrf_score
        FROM (
          SELECT chunk, rank_vec AS rank_val FROM embedding_recall
          UNION ALL
          SELECT chunk, rank_text AS rank_val FROM text_search_recall
        ) sub
        GROUP BY chunk
      ),
      top_chunks AS (
        SELECT chunk
        FROM rrf_scores
        ORDER BY rrf_score DESC
        LIMIT ' || rerank_recall_count || '
      ),
      concat_top_chunks AS (
        SELECT string_agg(chunk, E''\n\n----\n\n'') AS merged_chunks FROM top_chunks
      )
      SELECT ' || ai_gen_expr || '
      FROM concat_top_chunks;
    ';

    EXECUTE sql INTO final_answer;
    RETURN final_answer;
END;
$$ LANGUAGE plpgsql;