用亚马逊 Bedrock Agent 构建智能体和知识库 – 金蝶发票云票据助手开发实践（第二篇：企业私域问答开发实践）

目前，基于大语言模型的应用开发通常集中在两个方向：检索增强生成（RAG）和代理（Agent）。但无论是哪个方向，构建基于大模型的应用仅依赖 FM（基础模型）还远远不够，我们还需要集成多种组件，比如需要多种类型的数据库以便保存历史会话、向量数据；需要引入基于 FM 的应用开发框架（如 Langchain）简化开发；需要构建并优化提示词来保证 Agent Tools 准确调用，这无形中提升了开发门槛，拖慢了应用的上线步伐。

亚马逊在刚刚结束的 re:Invent 2023 发布大会上新推出的 Bedrock Agent 同时提供了 RAG 和 Agent 两种能力，通过提示词的自动构建、内置的历史会话管理、与 Lambda、多种向量库的自动集成，极大加速了企业用户对生成式 AI 的应用程序开发进程。Bedrock Agent 帮助用户更加专注于自身的业务，基于自身公司 API 和知识库来快速构建企业专用的智能体应用。

有关 Bedrock Agent 原理/实现的博客已有官方发布，本博客从代码、实操层面出发，带领大家基于 Bedrock Agent 从 0 到 1 快速实现智能体构建。场景是金蝶发票云探索基于基础大模型构建面向企业用户的发票助手，希望从以前用户需要面向系统做填单操作来开发票转变为通过自然语言对话申请发票开具。该 Agent 可以实现如下两个功能：

向客户收集开发票所需要的信息，自动调用后端 API 为用户开具发票
基于企业票据知识回答用户提问

本博客是系列博客的第二部分。将带领大家使用 Bedrock Knowledge Base（简称KB）的原生功能快速构建企业知识库；之后还会针对 QA 问答对场景，通过一步步的测试发现实际使用中遇到的问题，并逐步优化，实现知识库准确率的提升。

系列博客的第一部分，介绍如何基于 Bedrock Agent 构建发票开具助手的链接，请参见：https://aws.amazon.com/cn/blogs/china/build-agents-and-knowledge-bases-with-amazon-bedrock-agent-part-one/。

1. 在 Agent 中添加企业知识库

首先我们尝试向原有的 Agent 添加了一个“亚马逊员工手册”的 pdf 文档测试效果。测试过程非常简单：

在 Bedrock 中添加 KB
对 KB 进行描述
将包含 pdf 文件的 S3 存储桶目录作为 DataSource

在高级设置中我们可以根据文档的特性选择相应的文档切分方式
选择向量数据库存储向量索引
为简单起见，我们直接使用 Bedrock KB 帮我们自动创建的 Opensearch serverless 作为后端的向量数据库。
将 S3 中的数据同步至向量库
目前 Bedrock KB 可以支持的 Embeddings model 是 Titan Embeddings G1 – Text V1.2。在点击 sync 后后台就可以自动按照我们选择的文档切分方式进行文档切分、向量化、入 Opensearch serverless 向量库了。至此，一个基于 pdf 的文档向量库就构建完成了。
将知识库添加入博客一构建的 Agent 中
首先修改原有的 Agent Instruction，添加：我通过 KB“knowledge-base-amazonEmployee” 回答用户有关亚马逊员工手册的相关问题，以便 Agent 能够正确使用新加入的 KB。Agent Instrution：
您好，我是友好的发票助手。当收到问候时，我会礼貌地回复。我通过 Knowledge base“knowledge-base-amazonEmployee” 回答用户有关亚马逊员工手册的相关问题；我通过“issueInvoice” action group 提供开具发票服务。在生成发票时，我会首先从用户那里收集所有所需的发票信息。然后为用户生成临时预览图以供参考。如果成功生成了预览图像，我会从 generatePreviewInvoiceImage 函数的返回结果中获得发票金额、发票类型、购买方名称给用户。在生成实际发票之前，我会与用户确认是否要继续。如果用户确认，我会使用工具生成发票。如果成功，我会从函数结果中返回 downloadUrl 给用户，以便他们下载发票。如果用户表示信息不正确，我会要求他们提供更正后的信息。我还会确认用户是否需要把发票发送到指定的电子邮箱。如果要求发送,我会把发票文件发送到提供的邮箱。在 Agent 中加入前面创建的 KB
测试
经测试，效果不错。能准确地根据员工手册回答问题，并能继续完成开具发票的任务。

2. 将 QA 问答对加入到 KB

在将 pdf 文档成功加入到 agent 之后，我们再尝试将发票业务相关的 QA 问答对也加入到 Bedrock 的 KB 中。

企业中，实现智能客服通常是以 QA 问答对形式准备文档资料，如下为金蝶发票云知识库的示例片段。

Bedrock KB 可以支持的文档类型非常丰富，包括.pdf、.txt、.md、.html、.doc、.docx、.csv、.xls 和.xlsx 文件。上述的.xlsx 文件格式属于可以直接上传至 S3 进行处理的范围。于是我们将这个.xlsx 格式的 QA 问答对直接上传，尝试用章节一同样的方法构建一个知识库。

2.1 问题初现 — 没有使用 KB

但在测试中我们用知识库中的 Q 进行提问时发现了问题：

使用 Agent Test 中提供的原生的 trace 功能探究原因，我们看到 FM 根据用户的输入判断，该输入不属于 Agent 中任何一个 Action group 和 Knowledge Base 可以解决的问题，因此回复了“Sorry, I don’t have enough information to answer that.”。这说明我们的提示词表述出了问题。

2.2 通过修改提示词驱动 KB 的选择，但准确率低

为了处理这个问题我们将 Agent 和 KB 的指令都做再次修改，让 FM 更加清晰每个 KB 中包含的内容，从而做出正确的选择。

这里，我们总结了 QA 文档中提供的主要内容，并在 Agent Instruction 中做了描述：

测试后发现，这次 FM 能够正确的选择相应的 KB 了，并对知识库中内容做了召回。但回答的结果仍然不对。

再仔细追查原因，发现召回的内容与问题并不对应。

原来 Bedrock KB 原生提供的功能是将文档进行指定长度的切分。这种切分方式会将多个 QA 对合并到一起，导致召回内容多且不准确。

2.3 自行构建向量库，提升准确率

下面我们尝使用自行构建的向量库进一步解决这个问题。Bedrock KB 提供了两种向量库的构建方式。用户既可以选择 Quick create a new vector store 的方式让平台自动根据上传的内容创建 Opensearch serverless collection 用于承载上传的文档；也可以根据自己的需求自行完成向量库的构建。

为了提升对 QA 问答对的召回率，下面我们结合 Bedrock KB 的要求自行构建 Opensearch serverless collection，做如下优化：

将 QA 知识库进行逐条拆分，每个 item 仅包含一个 QA 问答对，减少无用内容的召回。
采用对称召回 + 非对称召回相结合的方式，分别对 Q 和 A 进行向量化，提升召回率。

下面我们展示实施过程涉及的部分关键代码：

在 Opensearch serverless 中构建”向量搜索”类型的向量集合

在该集合中创建用于保存知识库的 index
如下是参考 Bedrock KB 原生使用的 schema 创建的 index，必要的字段包含 Raw Text 文本字段、向量字段、documentID。
PUTpiaozone-qa-single

{
    "settings" : {
        "index":{
            "number_of_shards" : 1,
            "number_of_replicas" : 0,
            "knn": "true",
            "knn.algo_param.ef_search": 32
        }
    },
    "mappings": {
        "properties": {
        "AMAZON_BEDROCK_METADATA": {
          "type": "text",
          "index": false
        },
        "AMAZON_BEDROCK_TEXT_CHUNK": {
          "type": "text"
        },
        "bedrock-knowledge-base-default-vector": {
          "type": "knn_vector",
          "dimension": 1536,
          "method": {
            "engine": "nmslib",
            "space_type": "cosinesimil",
            "name": "hnsw",
            "parameters": {}
          }
        },
        "id": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        }
      }
    }
}

在后续处理中，我们分别把每个 QA 对做成两个 Document 插入 index，其中的 Answer 部分放入上述 index 中的 AMAZON_BEDROCK_TEXT_CHUNK 字段；同时分别对 Q 和 A 做向量化后放入 bedrock-knowledge-base-default-vector 字段。

将 QA 文档的每条 QA 对做向量化处理后插入到 index 中下面展示几段关键的代码段说明处理过程。

连接 Opensearch Serverless 向量库，通过加载到前面创建的 index 中。

 	def WriteVecIndexToAOS(bucket, object_key, content_type, smr_client, aos_endpoint=AOS_ENDPOINT, region=REGION,index_name=INDEX_NAME):
    credentials = boto3.Session().get_credentials()
    service = 'aoss'
    auth = AWS4Auth(credentials.access_key, credentials.secret_key,
                       region, service, session_token=credentials.token)
    try:
        file_content = load_content_json_from_s3(bucket, object_key, content_type, credentials)
        client = OpenSearch(
            hosts=[{'host': aos_endpoint, 'port': 443}],
            http_auth=auth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection,
            timeout=300,  # 默认超时时间是10 秒，
            max_retries=2,  # 重试次数
            retry_on_timeout=True
        )
        gen_aos_record_func = None
        if content_type in ["faq", "csv"]:
            gen_aos_record_func = iterate_QA(file_content, object_key, smr_client, index_name, EMB_MODEL_ENDPOINT)
        else:
            raise RuntimeError('No Such Content type supported')
        response = helpers.bulk(client, gen_aos_record_func, max_retries=3, initial_backoff=200, max_backoff=801,max_chunk_bytes=10 * 1024 * 1024)  # , chunk_size=10000, request_timeout=60000)
        return response
    except Exception as e:
        print(f"There was an error when ingest:{object_key} to aos cluster, Exception: {str(e)}")
        return None

逐条处理每个 QA 对，分别对 Q 进行向量化形成一条 Document；对 A 进行向量化再生成一条 Document，返回给 bulk 进行索引插入。

 	def iterate_QA(file_content, object_key, smr_client, index_name, endpoint_name):
    json_content = json.loads(file_content)
    json_arr = json_content["qa_list"]
    doc_title = object_key
    doc_category = json_content["doc_category"]
    print(json_arr)
    it = iter(json_arr)
    ##注：因为这里我们需要逐条处理每个QA对，因此将EMB_BATCH_SIZE设定为1。
    qa_batches = batch_generator(it, batch_size=EMB_BATCH_SIZE)

    doc_author = get_filename_from_obj_key(object_key)
    for idx, batch in enumerate(qa_batches):
        doc_template = "Answer: {}"
        questions = [item['Question'] for item in batch]
        print(questions)
        answers = [item['Answer'] for item in batch]
        docs = [doc_template.format(item['Answer']) for item in batch]
        authors = [item.get('Author') for item in batch]
        embeddings_q = get_embedding(smr_client, questions, endpoint_name)

        for i in range(len(embeddings_q)):
            document = {"AMAZON_BEDROCK_METADATA": """{"source":"s3://virginia199/piaozonecsv/piaozone.csv"}""",
                        "AMAZON_BEDROCK_TEXT_CHUNK": docs[i],
                        "bedrock-knowledge-base-default-vector": embeddings_q[i]}
            yield {"_index": index_name, "_source": document}

        embeddings_a = get_embedding(smr_client, answers, endpoint_name)
        for i in range(len(embeddings_a)):
            document = {"AMAZON_BEDROCK_METADATA": """{"source":"s3://virginia199/piaozonecsv/piaozone.csv"}""",
                        "AMAZON_BEDROCK_TEXT_CHUNK": docs[i],
                        "bedrock-knowledge-base-default-vector": embeddings_a[i],
                        "id": hashlib.md5(str(docs[i]).encode('utf-8')).hexdigest()}
            yield {"_index": index_name, "_source": document}