基于 Amazon Bedrock 托管 Claude3 实现 RAG 增强文本检索的 Text to SQL 方案

-- recruitment_data表 存储招聘岗位/面试流程/offer发放与反馈/入职相关字段
CREATE TABLE recruitment_data (
    application_id integer ENCODE az64, -- 主键 申请Id
    year integer ENCODE az64,
    time_type character varying(256) ENCODE lzo, -- 当前周 如 20240318-20240324
    application_time date ENCODE az64, -- 申请时间
    employment_title character varying(256) ENCODE lzo, -- 岗位名称/Id
    dept_id integer ENCODE az64, -- 招聘所属组织/部门的Id
    dept_name character varying(256) ENCODE lzo, -- 招聘所属组织/部门的名称
    interviewee_id character varying(256) ENCODE lzo, -- 面试者姓名/Id
    written_exercise_time timestamp without time zone ENCODE az64, -- 笔试时间
    written_exercise_is_need integer ENCODE az64,  -- 是否需要笔试
    written_exercise_pass integer ENCODE az64, -- 是否通过笔试
    first_interview_start_time timestamp without time zone ENCODE az64, -- 一面时间
    first_interview_is_need integer ENCODE az64,  -- 是否需要一面
    first_interview_pass integer ENCODE az64,  -- 是否通过一面
    second_interview_start_time date ENCODE az64,   -- 二面时间
    second_interview_is_need integer ENCODE az64,  -- 是否需要二面
    second_interview_pass integer ENCODE az64,  -- 是否通过二面
    hr_interview_start_time date ENCODE az64,  -- hr面时间
    hr_interview_is_need integer ENCODE az64,  -- 是否需要面
    hr_interview_pass integer ENCODE az64,  -- 是否通过hr面
    offer_time timestamp without time zone ENCODE az64, -- offer 发放时间
    interviewee_is_offered integer ENCODE az64, -- 面试者是否有offer
    interviewee_accept_offer integer ENCODE az64, -- 面试者是否接受offer
    interviewee_accept_offer_start_time timestamp without time zone ENCODE az64, -- offer 接受时间
    offer_feedback character varying(256) ENCODE lzo, -- offer 反馈内容
    interviewee_is_onboarded integer ENCODE az64, -- 候选人是否入职
    interviewee_onboarded_start_time timestamp without time zone ENCODE az64 -- 候选人入职时间
) DISTSTYLE AUTO;

-- dept_data表 存储部门名称 dept_data.dept_id与recruitment_data.dept_id 做外键关联
CREATE TABLE dept_data (
    dept_id integer ENCODE az64, -- 主键 部门Id
    dept_name character varying(256) ENCODE lzo, -- 部门名称
    l1_dept_id character varying(256) ENCODE lzo, -- 这个部门所属一集部门Id
    l2_dept_id character varying(256) ENCODE lzo, -- 这个部门所属二集部门Id
    l3_dept_id character varying(256) ENCODE lzo -- 这个部门所属三集部门Id
) DISTSTYLE AUTO;

你现在是一个AWS Redshift数据库查询助理, 现在有几张数据表需要你的查询，只返回一个你认为最正确的
表的名字, 不需要返回其他任何文本, 如果你认为没有对应的数据, 请返回
\"ERROR: You can only read data.\"。表如下：
'recruitment_data'表: 是招聘信息表, 存储招聘岗位和候选人投递、面试耗时、是否入职的相关信息,
不包含任何简历的信息。
'dept_data'表: 是部门信息表, 存储公司部门信息和组织架构。
你要查询的问题是：{被重复投递最多的是哪个岗位？}

// 一定要说明 你要查询的问题是，否则大语言模型不知道你问什么

// 角色设定
You are an expert in AWS Redshift. Your task is to understand the database tables if
given, and translate human instructions into SQL to query that database. I want you
to reply with a valid SQL in triple quotes. Do not write explanations. All valid
human instructions are given in curly braces {like this}. For any query that contains
delete or update in SQL, please respond: 'ERROR: You can only read data.'.
 涉及到比率等除法运算时，请将数据类型转换为 float4。 涉及到输出时间类型的运算时,
 请将数据转换成数值类型输出(秒数或者天数).
 
 // 表结构提示词
现在有一个aws redshift表, 表名是'recruitment_data', 存储的是一家公司的岗位招聘的应聘流程信息,
每一位应聘者每应聘一个岗位就会产生一条记录, 表内有如下列:'application_id', 数据类型是(integer), 
是主键; // ......其他列的描述，对于主要的字段要说明这一列含有那些数据，如employment_title包含['测试','开发','架构师']等值。
除了上述表结构, 生成SQL的时候, 还要参考以下公式:
offer接受率 = offer接受 / offer数:(coalesce({interviewee_accept_offer} * 1.0 /nullif({interviewee_is_offered}, 0), 0)),
SQL生成时的常见错误如下, 请在生成SQL的时候避免再犯这样的错误:
Redshift不会做隐式的数据类型转换, 因此如果数据类型需要转换, 其类型需要明确在SQL中指定。
有一类错误是对问题本身没有仔细的理解, 比如问题: “被重复投递最多的是哪个岗位？”,
可能会被错误的理解成”不同岗位的记录的重复数量“, 其真正的含义是需要理解到“重复”指的是候选人/应聘者
重复投递多个不同岗位, 或者一个岗位的情况。所以, 在理解问题的时候, 你需要考虑它的深层含义,
可以做适当的扩展和假设。
MUST output JSON, 仅生成SELECT语句。所有非计数的SQL必须限制结果不超过20条,
统计数量的SQL结果不限制,
如果SQL查询的结果适合柱状图, 请返回需要展示的字段和BarChart,
如果SQL查询的结果适合饼图, 请返回需要展示的字段和PieChart。
其中'columnList'属性是需要展示的字段的数组, 'chartType'属性是图表类型(BarChart,PieChart),
'finalSQL'属性是查询的SQL,
下面的例子你可以参考：

// RAG检索结果...
输出json格式示例：{query:"query",finalSQL:"finalSQL",chartType:"PieChart"}
你要查询的问题是：{被重复投递最多的是哪个岗位？}

// 一定要说明 你要查询的问题是，否则大语言模型不知道你问什么

现在您需要对数据结果进行分析, 并尽可能的对指标的变化作出下钻归因, 数据受众xxxxx,
添加您对HR招聘业务的理解,不要暴露具体的SQL字段。
数据是：
// 序列化后的数据
存储他的数据库表结构是 // 第二步中prompt设置的表schema

把它查询出来的SQL是：// 第二步中返回的SQL
Query: {被重复投递最多的是哪个岗位？请根据数据做智能分析}

// 您也可以在此添加背景信息，方便大语言模型根据背景信息做下钻分析。

// 第二次Prompt原文
// 增加
1.请注意, 在生成SQL的时候, 这里有几个追加的要求, 具体如下: 先检查问题的句子成分和含义成分是否清晰,
是否有歧义, 如果不清晰的话要进行扩展, 使问题变得清晰。将澄清后的问题输出到'clarify'属性中。
你需要一步一步思考, 并对SQL进行解释, 解释部分放在输出的'reasoning'属性中。
针对一个问题, 由于SQL可能有不同的写法, 你需要先从不同的角度思考,生成最多五个针对当前问题不同的写法
的SQL, 并包含独立的解释。生成的SQL结果放到'referenceSql'属性中。你接下来要检查上面的几个SQL是
否有错误, 检查的时候你要从是否正确理解问题等角度, 重新思考, 并假定SQL就是错误的。
检查的结果需要写到每一个SQL的'check1'和'check2'...'checkN'属性中。如果没有错误就写它全对的概率,
如{'check1': {'no_error': 0.6'}}, 如果有错误则参考这样的输出例子:
 {'check2':{'error': 'VARCHAR需要转换成Float类型。'}}
 最后一步是根据上面的几个'referenceSql'和各自'check'的结果confidence来生成你认为正确的最终SQL
 和分析, 放到'finalSQL', 'reasoningFinal'中。
 2. 结果样式的例子如下（注意顺序）: // 一定要加注意顺序 否则不会触发自检查
 {
    'clarify': '问题应该是: XYZ ',
    'reasoning1': 'reason AAA',
    'referenceSql1': 'sql AAA',
    'reasoning2': 'reason BBB',
    'referenceSql2': 'sql BBB',
    'check1': 'For SQL AAA: no error',
    'check2': 'For SQL BBB: VARCHAR需要转换成Float类型',
    'reasoningFinal': 'reason for final SQL',
    'finalSQL': 'The final SQL based on above two reference sql and the error check result.',
    'columnList': ['column1', 'column2']
}

import {
  BedrockRuntimeClient,
  InvokeModelCommand,
} from '@aws-sdk/client-bedrock-runtime';

const bedrockStream = async (
  messages: Message[],
  promptConfig: any,
  requestId: string
) => {
  try {
    const AWS_PARAM = {
        region:"",
        credentials:{}
    }
    const bedrockruntime = new BedrockRuntimeClient(AWS_PARAM);

    const userQuestion = messages[messages.length - 1].content;
    const promptStr = `各类提示词 RAG检索结果${promptConfig} 你要回答的问题是: {${userQuestion}}`;
    
    const bedrockResult = await bedrockClaude3Respone(
      [
        ...(delete messages[messages.length-1]),
        {
          role: "user",
          content: promptStr,
        },
      ],
      bedrockruntime,
      requestId
    );
    const bedrockResultObj = JSON.parse(bedrockResult);
    return bedrockResultObj;

    // 后续再解析SQL查询DB/Redshift ，进行Insight
  } catch (error) {
    console.error(
      "Bedrock request or respone error " + requestId + " >>>",
      error
    );
    throw new Error(`Bedrock request or respone error`);
  }
};

const bedrockClaude3Respone = async (
  bodyData: any[],
  bedrockruntime: {
    send: (
      arg0: InvokeModelWithResponseStreamCommand | InvokeModelCommand,
    ) => any;
  },
  requestId: string,
) => {

  
  const command = new InvokeModelCommand({
    body: JSON.stringify({
      anthropic_version: 'bedrock-2023-05-31',
      messages: bodyData,
      max_tokens: BEDROCK_MAX_TOKEN,
    }),
    modelId: 'anthropic.claude-3-sonnet-20240229-v1:0',
    contentType: 'application/json',
    accept: 'application/json',
  });
  const respone = await bedrockruntime.send(command);
  if (respone.$metadata.httpStatusCode !== 200) {
    console.error('Bedrock respone error >>>', respone);
    throw new Error(`Bedrock returned an error`);
  }
  const resultData = await resolveClaude3Body(respone.body, requestId);
  return resultData;
};

const resolveClaude3Body = async (
  responeBody: any,
  requestId: string | undefined,
) => {
  // 解析内容
  const resultStr = responeBody.transformToString('utf8');
  const resultJson = JSON.parse(resultStr);
  return resultJson.content[0].text;
};