最佳实践：金融多模态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;
  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 := NULL;
          corpus_name   := corpus_table;
      END IF;
  
      obj_table_name := corpus_name || '_obj_table';
  
      IF corpus_schema IS NULL THEN
          full_corpus_ident := format('%I', corpus_name);
          full_obj_ident    := format('%I', obj_table_name);
      ELSE
          full_corpus_ident := format('%I.%I', corpus_schema, corpus_name);
          full_obj_ident    := format('%I.%I', corpus_schema, obj_table_name);
      END IF;
      
      -- 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 = COALESCE(corpus_schema, 'public')
              )
              INTO dyn_table_exists;
  
              IF dyn_table_exists THEN
                  -- 2.1 关闭动态表自动刷新
                  RAISE NOTICE 'Disabling auto refresh for %', full_corpus_ident;
                  EXECUTE format(
                      'ALTER TABLE %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)';
      ELSE
          embed_expr := format(
              'ai_embed(%L, chunk)',
              embedding_model
          );
      END IF;
  
      -- 8. 创建 RAG 输出动态表
      RAISE NOTICE 'create dynamic table: %', corpus_name;
      EXECUTE format(
          $f$
          CREATE DYNAMIC TABLE %s
          WITH (
              auto_refresh_enable = FALSE ,
              auto_refresh_mode = 'incremental',
              freshness = '5 minutes'
          ) 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,
          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 := NULL;
          corpus_name   := corpus_table;
      END IF;
  
      obj_table_name := corpus_name || '_obj_table';
  
      IF corpus_schema IS NULL THEN
          full_corpus_ident := format('%I', corpus_name);
          full_obj_ident    := format('%I', obj_table_name);
      ELSE
          full_corpus_ident := format('%I.%I', corpus_schema, corpus_name);
          full_obj_ident    := format('%I.%I', corpus_schema, obj_table_name);
      END IF;
  
      -- 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 := NULL;
          corpus_name   := corpus_table;
      END IF;
  
      obj_table_name := corpus_name || '_obj_table';
  
      IF corpus_schema IS NULL THEN
          full_corpus_ident := format('%I', corpus_name);
          full_obj_ident    := format('%I', obj_table_name);
      ELSE
          full_corpus_ident := format('%I.%I', corpus_schema, corpus_name);
          full_obj_ident    := format('%I.%I', corpus_schema, obj_table_name);
      END IF;
  
      -- 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 Object 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_retrieval(
    question TEXT,
    vector_index TEXT,
    embedding_model TEXT DEFAULT NULL,
    llm_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT '请根据参考信息回答以下问题：${question}\n\n 参考信息：\n\n ${context}',
    vector_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    embedding_table TEXT;
    embedding_col TEXT;
    sql TEXT;
    embedding_expr TEXT;
    ai_rank_expr TEXT;
    ai_gen_expr TEXT;
    llm_model_valid BOOLEAN;
BEGIN
    -- 解析 vector_index
    SELECT table_path, column_name
    INTO embedding_table, embedding_col
    FROM parse_index(vector_index);

    -- embedding表达式处理
    IF embedding_model IS NULL OR trim(embedding_model) = '' THEN
        embedding_expr := 'ai_embed(questions.query_text)';
    ELSE
        embedding_expr := 'ai_embed(' || quote_literal(embedding_model) || ', questions.query_text)';
    END IF;

    -- LLM模型处理
    llm_model_valid := (llm_model IS NOT NULL AND trim(llm_model) != '');

    -- ai_rank表达式
    IF llm_model_valid THEN
        ai_rank_expr := 'ai_rank(' || quote_literal(llm_model) || ', questions.query_text, chunk)';
    ELSE
        ai_rank_expr := 'ai_rank(questions.query_text, chunk)';
    END IF;

    -- ai_gen表达式：通过模板替换生成完整 prompt
    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(' || quote_literal(prompt) || ', ''${question}'', questions.query_text), ''${context}'', merged_chunks) )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(' || quote_literal(prompt) || ', ''${question}'', questions.query_text), ''${context}'', merged_chunks))';
    END IF;

    -- 动态SQL
    sql := '
    WITH
      questions AS (
        SELECT ' || quote_literal(question) || ' AS query_text
      ),
      embedding_recall AS (
        SELECT
          chunk,
          approx_cosine_distance(' || embedding_col || ', ' || embedding_expr || ') AS distance
        FROM
          ' || embedding_table || ', questions
        ORDER BY
          distance DESC
        LIMIT ' || vector_recall_count || '
      ),
      rerank AS (
        SELECT
          chunk,
          ' || ai_rank_expr || ' AS score
        FROM
          embedding_recall, questions
        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, questions;
    ';

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

全文检索存储过程

CREATE OR REPLACE FUNCTION qa_text_search_retrieval(
    question TEXT,
    text_search_index TEXT,
    llm_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT '请根据参考信息回答以下问题：${question}\n\n参考信息：\n\n${context}',
    text_search_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    text_search_table TEXT;
    text_search_col TEXT;
    sql TEXT;
    ai_rank_expr TEXT;
    ai_gen_expr TEXT;
    llm_model_valid BOOLEAN;
BEGIN
    -- 解析 text_search_index
    SELECT table_path, column_name
    INTO text_search_table, text_search_col
    FROM parse_index(text_search_index);

    -- LLM 模型处理
    llm_model_valid := (llm_model IS NOT NULL AND trim(llm_model) != '');

    -- ai_rank 表达式
    IF llm_model_valid THEN
        ai_rank_expr := 'ai_rank(' || quote_literal(llm_model) || ', questions.query_text, chunk)';
    ELSE
        ai_rank_expr := 'ai_rank(questions.query_text, chunk)';
    END IF;

    -- ai_gen 表达式
    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(' || quote_literal(prompt) || 
              ', ''${question}'', questions.query_text), ''${context}'', merged_chunks) )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(' || quote_literal(prompt) || 
              ', ''${question}'', questions.query_text), ''${context}'', merged_chunks))';
    END IF;

    -- 动态 SQL
    sql := '
    WITH
      questions AS (
        SELECT ' || quote_literal(question) || ' AS query_text
      ),
      -- 全文搜索召回
      text_search_recall AS (
        SELECT
          chunk
        FROM
          ' || text_search_table || ', questions
        ORDER BY
          text_search(' || text_search_col || ', ' || quote_literal(question) || ') DESC
        LIMIT ' || text_search_recall_count || '
      ),
      -- rerank
      rerank AS (
        SELECT
          chunk,
          ' || ai_rank_expr || ' AS score
        FROM
          text_search_recall, questions
        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, questions;
    ';

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

向量+全文双路召回+Rank排序存储过程

-- 全文、向量双路召回 + rerank
CREATE OR REPLACE FUNCTION qa_hybrid_retrieval(
    question TEXT,
    vector_index TEXT,
    text_search_index TEXT,
    embedding_model TEXT DEFAULT NULL,
    llm_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT '请根据参考信息回答以下问题：${question}\n\n 参考信息：\n\n ${context}',
    text_search_recall_count INT DEFAULT 20,
    vector_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    embedding_table TEXT;
    embedding_col TEXT;
    text_search_table TEXT;
    text_search_col TEXT;
    sql TEXT;
    embedding_expr TEXT;
    ai_rank_expr TEXT;
    ai_gen_expr TEXT;
    llm_model_valid BOOLEAN;
BEGIN
    -- 解析 vector_index
    SELECT table_path, column_name
    INTO embedding_table, embedding_col
    FROM parse_index(vector_index);

    -- 解析 text_search_index
    SELECT table_path, column_name
    INTO text_search_table, text_search_col
    FROM parse_index(text_search_index);

    -- embedding表达式处理
    IF embedding_model IS NULL OR trim(embedding_model) = '' THEN
        embedding_expr := 'ai_embed(questions.query_text)';
    ELSE
        embedding_expr := 'ai_embed(' || quote_literal(embedding_model) || ', questions.query_text)';
    END IF;

    -- LLM模型处理
    llm_model_valid := (llm_model IS NOT NULL AND trim(llm_model) != '');

    -- ai_rank表达式
    IF llm_model_valid THEN
        ai_rank_expr := 'ai_rank(' || quote_literal(llm_model) || ', questions.query_text, chunk)';
    ELSE
        ai_rank_expr := 'ai_rank(questions.query_text, chunk)';
    END IF;

    -- ai_gen表达式：通过模板替换生成完整 prompt
    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(' || quote_literal(prompt) || ', ''${question}'', questions.query_text), ''${context}'', merged_chunks) )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(' || quote_literal(prompt) || ', ''${question}'', questions.query_text), ''${context}'', merged_chunks))';
    END IF;

    -- 构造动态SQL（双路召回 union 后再 rerank）
    sql := '
    WITH
      questions AS (
        SELECT ' || quote_literal(question) || ' AS query_text
      ),
      -- 向量召回
      embedding_recall AS (
        SELECT
          chunk
        FROM
          ' || embedding_table || ', questions
        ORDER BY
          approx_cosine_distance(' || embedding_col || ', ' || embedding_expr || ') DESC
        LIMIT ' || vector_recall_count || '
      ),
      -- 全文搜索召回
      text_search_recall AS (
        SELECT
          chunk
        FROM
          ' || text_search_table || ', questions
        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
      rerank AS (
        SELECT
          chunk,
          ' || ai_rank_expr || ' AS score
        FROM
          union_recall, questions
        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, questions;
    ';

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

向量+全文双路召回+RRF排序存储过程

-- 全文、向量双路召回 + RRF rerank (Reciprocal Rank Fusion)
CREATE OR REPLACE FUNCTION qa_hybrid_retrieval_rrf(
    question TEXT,
    vector_index TEXT,
    text_search_index TEXT,
    embedding_model TEXT DEFAULT NULL,
    llm_model TEXT DEFAULT NULL,
    prompt TEXT DEFAULT '请根据参考信息回答以下问题：${question}\n\n 参考信息：\n\n ${context}',
    text_search_recall_count INT DEFAULT 20,
    vector_recall_count INT DEFAULT 20,
    rerank_recall_count INT DEFAULT 5,
    rrf_k INT DEFAULT 60
)
RETURNS TEXT AS
$$
DECLARE
    final_answer TEXT;
    embedding_table TEXT;
    embedding_col TEXT;
    text_search_table TEXT;
    text_search_col TEXT;
    sql TEXT;
    embedding_expr TEXT;
    ai_gen_expr TEXT;
    llm_model_valid BOOLEAN;
BEGIN
    -- 解析 vector_index
    SELECT table_path, column_name
    INTO embedding_table, embedding_col
    FROM parse_index(vector_index);

    -- 解析 text_search_index
    SELECT table_path, column_name
    INTO text_search_table, text_search_col
    FROM parse_index(text_search_index);

    -- 构造 embedding 调用表达式
    IF embedding_model IS NULL OR trim(embedding_model) = '' THEN
        embedding_expr := 'ai_embed(questions.query_text)';
    ELSE
        embedding_expr := 'ai_embed(' || quote_literal(embedding_model) || ', questions.query_text)';
    END IF;

    -- 检查是否指定了 LLM 模型
    llm_model_valid := (llm_model IS NOT NULL AND trim(llm_model) != '');

    -- 构造 ai_gen 调用表达式（替换 prompt 模板里的变量）
    IF llm_model_valid THEN
        ai_gen_expr := 'ai_gen(' || quote_literal(llm_model) ||
            ', replace(replace(' || quote_literal(prompt) ||
            ', ''${question}'', questions.query_text), ''${context}'', merged_chunks) )';
    ELSE
        ai_gen_expr := 'ai_gen(replace(replace(' || quote_literal(prompt) ||
            ', ''${question}'', questions.query_text), ''${context}'', merged_chunks))';
    END IF;

    -- 拼 SQL
    sql := '
    WITH
      questions AS (
        SELECT ' || quote_literal(question) || ' AS query_text
      ),

      -- 向量检索召回 + 排名（approx_cosine_distance 调用一次）
      embedding_recall AS (
        SELECT
          chunk,
          vec_score,
          ROW_NUMBER() OVER (ORDER BY vec_score DESC) AS rank_vec
        FROM (
          SELECT
            chunk,
            approx_cosine_distance(' || embedding_col || ', ' || embedding_expr || ') AS vec_score
          FROM
            ' || embedding_table || ', questions
        ) t
        ORDER BY vec_score DESC
        LIMIT ' || vector_recall_count || '
      ),

      -- 全文检索召回 + 排名（score 已有，不重复算）
      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
            ' || text_search_table || '
        ) ts
        WHERE text_score > 0
        ORDER BY text_score DESC
        LIMIT ' || text_search_recall_count || '
      ),

      -- 计算 RRF 分数
      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 N
      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, questions;
    ';

    -- 执行并返回
    EXECUTE sql INTO final_answer;
    RETURN final_answer;
END;
$$ LANGUAGE plpgsql;