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

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

代码结构：

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

代码样例：

注：添加chunk_attachment字段返回的视频关键帧抽取图列后会对回答的耗时和token消耗有较大增加，可先使用返回字段中的content和description内容测试问答效果，若符合问答要求可添加视频关键帧抽取图列（注意调小limit，防止超过大模型问答窗口）

import json
import requests
import datetime

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

collection_name = ""
project_name = ""
query = ""
ak = ""
sk = ""
account_id = ""
g_knowledge_base_domain = ""

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

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

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

# 参考资料

注意：「参考资料」可以为文本、图片、视频等多种内容
- 文本资料是一段文本
- 图片资料则是图片内容，可能会包括关于图片的描述性文本
- 视频内容则是由一组图片组成，这组图片是由一段视频按顺序抽帧得到的，每一个视频内容可能剪切自同一个视频也可能剪切自不同的视频
<context>
  {}
</context>
参考资料中提到的图和视频按上传顺序排列，请结合图片、视频与文本信息综合回答问题。如参考资料中没有图片或视频，请仅根据参考资料中的文本信息回答问题。

# 引用要求
1. 当可以回答时，在句子末尾适当引用相关参考资料，每个参考资料引用格式必须使用<reference>标签对，例如: <reference data-ref=\"{{point_id}}\"></reference>。
2. 当告知客户无法回答时，不允许引用任何参考资料。
3. \"data-ref\" 字段表示对应参考资料的 point_id。
4. 适当合并引用，当引用项相同可以合并引用，只在引用内容结束添加一个引用标签。

# 视频插入
1. 在适当的时候，可以在回答中合适的位置插入视频内容，来增加回答的丰富度，插入视频的方法如下所示。
2. 首先对参考资料的每个视频内容含义深入理解，然后从所有视频内容中筛选出与回答上下文直接关联的视频，在回答中的合适位置插入，视频内容必须支持直接的可视化说明问题的答案。若不需要插入视频，则省略视频的插入。
3. 使用 <video> 标签对表示插入视频，例如: <video data-ref=\"{{point_id}}\"></video>，其中 \"point_id\" 字段表示对应视频的 point_id，每个新插入的视频标签对必须另起一行，相同的视频（以\"point_id\"区分）只允许使用一次。
"""

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,
        "return_token_usage": True,
        "messages": [
            {
                "role": "system",
                "content": ""
            },
            {
                "role": "user",
                "content": ""
            }
        ],
        "rewrite": False
    },
    "post_processing": {
        "get_attachment_link": True,
        "rerank_only_chunk": False,
        "rerank_switch": False,
        "chunk_group": True,
        "chunk_diffusion_count": 0
    },
    "dense_weight": 0.5
}

    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

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": True,
        "return_token_usage": True,
        "model": "Doubao-seed-1-6",
        "max_tokens": 32768,
        "temperature": 1,
        "model_version": "251015",
        "thinking": {
             "type":"enabled"
        }
    }

    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
    )
    rsp.encoding = "utf-8"
    print("chat completion res = {}".format(rsp.text))

def is_vision_model(model_name,model_version):
    if model_name is None:
        return False
    return "vision" in model_name or "m" in model_version or "seed" in model_name
def get_content_for_prompt(point: dict) -> str:
    content = point["content"]
    original_question = point.get("original_question")
    if original_question:
        # faq 召回场景，content 只包含答案，需要把原问题也拼上
        return "当询问到相似问题时，请参考对应答案进行回答：问题：“{question}”。答案：“{answer}”".format(
                question=original_question, answer=content)
    return content

def generate_prompt(rsp_txt):
    rsp = json.loads(rsp_txt)
    if rsp["code"] != 0:
        return "", []
    prompt = ""
    rsp_data = rsp["data"]
    points = rsp_data["result_list"]
    using_vlm = is_vision_model("Doubao-seed-1-6","251015")
    content = []

    for point in points:
        doc_text_part = ""
        # 先拼接系统字段
        doc_info = point["doc_info"]
        for system_field in ["point_id","doc_name","title","chunk_title","content","video_start_time","video_end_time"] : 
            if system_field == 'doc_name' or system_field == 'title':
                if system_field in doc_info:
                    doc_text_part += f"{system_field}: {doc_info[system_field]}\n"
            else:
                if system_field in point:
                    if system_field == "content":
                        doc_text_part += f"content: {get_content_for_prompt(point)}\n"
                    elif system_field == "point_id":
                        doc_text_part += f"point_id: \"{point['point_id']}\""
                    elif system_field == "video_start_time" or system_field == "video_end_time":
                        formatted_time = datetime.datetime.fromtimestamp(int(point[system_field]) / 1000).strftime("%H:%M:%S")
                        doc_text_part += f"{system_field}: {formatted_time}\n"
                    else:
                        doc_text_part += 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:
                    doc_text_part += f"{self_field}: {find_one['field_value']}\n"

        # 提取图片链接
        image_list =[]
        if using_vlm and "chunk_attachment" in point:
            for attachment in point["chunk_attachment"]:
                image_list.append(attachment["link"])
            if image_list:
                prompt += f"视频抽帧内容如下: \n"

        
        content.append({
            'type': 'text',
            'text': doc_text_part
        })
        if image_list:
            for image_link in image_list:
                content.append({
                    'type': 'image_url',
                    'image_url': {
                        'url': image_link
                    }
                })
        prompt += f"{doc_text_part}\n"

    if using_vlm:
        content_pre_sub = base_prompt.split('{}')
        content_pre = {'type': 'text', 'text': content_pre_sub[0]}
        content_sub = {'type': 'text', 'text': content_pre_sub[1]}
        return [content_pre] + content + [content_sub]
    else:
        return base_prompt.format(prompt)

def search_knowledge_and_chat_completion():
    # 1.执行search_knowledge
    rsp_txt = search_knowledge()
    # 2.生成prompt
    prompt = generate_prompt(rsp_txt)
    # todo:用户需要本地缓存对话信息，并按照顺序依次加入到messages中
    # 3.拼接message对话, 问题对应role为user，系统对应role为system, 答案对应role为assistant, 内容对应content
    user_content = query
    messages = [
        {
            "role": "system",
            "content": prompt
        },
        {
            "role": "user",
            "content": user_content
        }
    ]

    # 4.调用chat_completion
    chat_completion(messages)

if __name__ == "__main__":
    search_knowledge_and_chat_completion()