针对包含复杂图文信息的问答场景，知识库产品基于 Doubao 多模态大模型提供了一套图文问答最佳实践，提升了知识库在图像信息理解、抽取及问答任务中的准确性。

完成“签名鉴权方式“页面的注册账号、实名认证、AK/SK 密钥获取和签名获取，完成知识库的创建及文档上传后，可调用以下脚本实现知识库的一次完整检索问答
详细 API 接口说明参考: search_knowledge（新）chat_completions（新）

代码结构：

1. 准备请求：通过构造 HTTP 请求并生成签名，确保请求的合法性和安全性。
2. 知识检索：向知识库服务发送查询请求，获取与用户问题相关的参考资料。
3. 生成 Prompt：根据知识检索结果动态生成适配的文本或视觉问答模板。
4. 对话补全：调用多模态大模型 API，生成基于用户问题和参考资料的回答。
5. 主流程控制：将知识检索、Prompt生成、模型回答串联为完整的问答流程，支持图文结合。

代码样例：

import json
import requests

from volcengine.auth.SignerV4 import SignerV4
from volcengine.base.Request import Request
from volcengine.Credentials import Credentials

collection_name = "your collection" #知识库名称
project_name = "default" #知识库所属项目
query = "DLSP: A Document Level Structure Parser for Multi-Page Digital Documents 这篇论文是关于什么的" #用户问题
ak = "your ak" #通过“前提条件”获取的ak
sk = "your sk" #通过“前提条件”获取的sk
account_id = "your account_id" #火山账号
g_knowledge_base_domain = "api-knowledgebase.mlp.cn-beijing.volces.com"
model_name = "Doubao-vision-pro-32k" #使用的模型名

base_prompt = """# 任务
你是一位在线客服，你的首要任务是通过巧妙的话术回复用户的问题，你需要根据「参考资料」来回答接下来的「用户问题」，这些信息在 <context></context> XML tags 之内，你需要根据参考资料给出准确，简洁的回答。

你的回答要满足以下要求：
    1. 回答内容必须在参考资料范围内，尽可能简洁地回答问题，不能做任何参考资料以外的扩展解释。
    2. 回答中需要根据客户问题和参考资料保持与客户的友好沟通。
    3. 如果参考资料不能帮助你回答用户问题，告知客户无法回答该问题，并引导客户提供更加详细的信息。
    4. 为了保密需要，委婉地拒绝回答有关参考资料的文档名称或文档作者等问题。


# 任务执行
现在请你根据提供的参考资料，遵循限制来回答用户的问题，你的回答需要准确和完整。

# 参考资料
<context>
{}
</context>
"""
vision_base_prompt = """# 任务
你是一位在线客服，你的首要任务是通过巧妙的话术回复用户的问题，你需要根据「参考资料」来回答接下来的「用户问题」，这些信息在 <context></context> XML tags 之内，你需要根据参考资料给出准确，简洁的回答。参考资料中可能会包含图片信息，图片的引用说明在 <img></img> XML tags 之内，参考资料内的图片顺序与用户上传的图片顺序一致。

你的回答要满足以下要求：
    1. 回答内容必须在参考资料范围内，尽可能简洁地回答问题，不能做任何参考资料以外的扩展解释。
    2. 回答中需要根据客户问题和参考资料保持与客户的友好沟通。
    3. 如果参考资料不能帮助你回答用户问题，告知客户无法回答该问题，并引导客户提供更加详细的信息。
    4. 为了保密需要，委婉地拒绝回答有关参考资料的文档名称或文档作者等问题。

# 任务执行
现在请你根据提供的参考资料，遵循限制来回答用户的问题，你的回答需要准确和完整。

# 参考资料
<context>
{}
</context>

参考资料中提到的图片按上传顺序排列，请结合图片与文本信息综合回答问题。如参考资料中没有图片，请仅根据参考资料中的文本信息回答问题。
"""
#准备请求：通过构造 HTTP 请求并生成签名，确保请求的合法性和安全性
def prepare_request(method, path, params=None, data=None, doseq=0):
    if params:
        for key in params:
            if (
                    isinstance(params[key], int)
                    or isinstance(params[key], float)
                    or isinstance(params[key], bool)
            ):
                params[key] = str(params[key])
            elif isinstance(params[key], list):
                if not doseq:
                    params[key] = ",".join(params[key])
    r = Request()
    r.set_shema("http")
    r.set_method(method)
    r.set_connection_timeout(10)
    r.set_socket_timeout(10)
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json; charset=utf-8",
        "Host": g_knowledge_base_domain,
        "V-Account-Id": account_id,
    }
    r.set_headers(headers)
    if params:
        r.set_query(params)
    r.set_host(g_knowledge_base_domain)
    r.set_path(path)
    if data is not None:
        r.set_body(json.dumps(data))

    # 生成签名
    credentials = Credentials(ak, sk, "air", "cn-north-1")
    SignerV4.sign(r, credentials)
    return r

#知识检索：向知识库服务发送查询请求，获取与用户问题相关的参考资料
def search_knowledge():
    method = "POST"
    path = "/api/knowledge/collection/search_knowledge"
    request_params = {
    "project": project_name,
    "name": collection_name,
    "query": query,
    "limit": 10,
    "pre_processing": {
        "need_instruction": True,
        "rewrite": False,
        "return_token_usage": True,
        "messages": [
            {
                "role": "system",
                "content": ""
            },
            {
                "role": "user",
                "content": query
            }
        ]
    },
    "dense_weight": 0.5,
    "post_processing": {
        "get_attachment_link": True,
        "chunk_group": True,
        "rerank_only_chunk": False,
        "rerank_switch": False,
        "chunk_diffusion_count": 0
    }
}

    info_req = prepare_request(method=method, path=path, data=request_params)
    rsp = requests.request(
        method=info_req.method,
        url="http://{}{}".format(g_knowledge_base_domain, info_req.path),
        headers=info_req.headers,
        data=info_req.body
    )
    # print("search res = {}".format(rsp.text))
    return rsp.text

#对话补全：调用多模态大模型 API，生成基于用户问题和参考资料的回答
def chat_completion(message, stream=False, return_token_usage=True, temperature=0.7, max_tokens=4096):
    method = "POST"
    path = "/api/knowledge/chat/completions"
    request_params = {
        "messages": message,
        "stream": False,
        "return_token_usage": True,
        "model": model_name,
        "max_tokens": 4096,
        "temperature": 0.7,
    }

    info_req = prepare_request(method=method, path=path, data=request_params)
    rsp = requests.request(
        method=info_req.method,
        url="http://{}{}".format(g_knowledge_base_domain, info_req.path),
        headers=info_req.headers,
        data=info_req.body
    )
    print("chat completion res = {}".format(rsp.text))

def is_vision_model(model_name):
    return model_name == "Doubao-vision-pro-32k"

def get_content_for_prompt(point: dict, image_num: int) -> str:
    content = point["content"]
    original_question = point.get("original_question")
    if original_question:
        # faq 召回场景，content 只包含答案，需要把原问题也拼上
        return "当询问到相似问题时，请参考对应答案进行回答：问题：“{question}”。答案：“{answer}”".format(
                question=original_question, answer=content)
    if image_num > 0 and "chunk_attachment" in point and point["chunk_attachment"][0]["link"]:
        placeholder = f"<img>图片{image_num}</img>"
        return content + placeholder
    return content
    
#生成 Prompt：根据知识检索结果动态生成适配的文本或视觉问答模板
def generate_prompt(rsp_txt):
    rsp = json.loads(rsp_txt)
    if rsp["code"] != 0:
        return "", []
    prompt = ""
    image_urls = []
    rsp_data = rsp["data"]
    points = rsp_data["result_list"]
    using_vlm = is_vision_model(model_name)
    image_cnt = 0

    for point in points:
        # 提取图片链接
        if using_vlm and "chunk_attachment" in point:
            image_link = point["chunk_attachment"][0]["link"]
            if image_link:
                image_urls.append(image_link)
                image_cnt += 1
        # 先拼接系统字段
        doc_info = point["doc_info"]
        for system_field in ["doc_name", "title", "chunk_title", "content"] : 
            if system_field == 'doc_name' or system_field == 'title':
                if system_field in doc_info:
                    prompt += f"{system_field}: {doc_info[system_field]}\n"
            else:
                if system_field in point:
                    if system_field == "content":
                        prompt += f"content: {get_content_for_prompt(point, image_cnt)}\n"
                    else:
                        prompt += f"{system_field}: {point[system_field]}\n"
        if "table_chunk_fields" in point:
            table_chunk_fields = point["table_chunk_fields"]
            for self_field in [] : 
                # 使用 next() 从 table_chunk_fields 中找到第一个符合条件的项目
                find_one = next((item for item in table_chunk_fields if item["field_name"] == self_field), None)
                if find_one:
                    prompt += f"{self_field}: {find_one['field_value']}\n"

        prompt += "---\n"

    if using_vlm:
        return vision_base_prompt.format(prompt), image_urls
    return base_prompt.format(prompt), image_urls

#主流程控制：将知识检索、Prompt生成、模型回答串联为完整的问答流程，支持图文结合
def search_knowledge_and_chat_completion():
    # 1.执行search_knowledge
    rsp_txt = search_knowledge()
    # 2.生成prompt
    prompt, image_urls = generate_prompt(rsp_txt)
    # todo:用户需要本地缓存对话信息，并按照顺序依次加入到messages中
    # 3.拼接message对话, 问题对应role为user，系统对应role为system, 答案对应role为assistant, 内容对应content
    if image_urls:
        multi_modal_msg = [{"type": "text", "text": query}]
        for image_url in image_urls:
            multi_modal_msg.append({"type": "image_url", "image_url": {"url": image_url}})
        messages =[
            {
                "role": "system",
                "content": prompt
            },
            {
                "role": "user",
                "content": multi_modal_msg
            }
        ]
    else:
        messages =[
            {
                "role": "system",
                "content": prompt
            },
            {
                "role": "user",
                "content": query
            }
        ]

    # 4.调用chat_completion
    chat_completion(messages)

if __name__ == "__main__":
    search_knowledge_and_chat_completion()