核心流程--向量数据库VikingDB-火山引擎

文档中心

立即注册

向量数据库VikingDB

请输入

Python SDK

核心流程

本页面提供一个通过 Python SDK 完成知识库文档导入与知识问答的完整流程请求示例，涵盖文档上传、向量检索及大模型问答调用，注意使用时根据实际情况填充账户鉴权信息、collection 名称、文档路径和查询内容。

import os
import sys
from vikingdb.auth import IAM
from vikingdb.knowledge import VikingKnowledge
from vikingdb.knowledge.exceptions import VikingKnowledgeException
from vikingdb.knowledge.models.doc import AddDocV2Request
from vikingdb.knowledge.models.search import SearchKnowledgeRequest
from vikingdb.knowledge.models.chat import ChatMessage, ChatCompletionRequest, ChatCompletionResponse



def main():
    access_key = os.environ["VIKINGDB_AK"]
    secret_key = os.environ["VIKINGDB_SK"]


    endpoint = "api-knowledgebase.mlp.cn-beijing.volces.com"
    region = "cn-beijing"


    collection_name = "your-collection-name"
    project_name = "default"


    client = VikingKnowledge(
        host=endpoint,
        region=region,
        auth=IAM(ak=access_key, sk=secret_key),
        scheme="https",
    )


    collection = client.collection(
        collection_name=collection_name,
        project_name=project_name,
    )


    tos_uri = "tos://your-bucket/your-path/your-file.pdf"
    rsp = collection.add_doc_v2(
        AddDocV2Request(
            doc_id="your-doc-id",
            uri=tos_uri,
        )
    )
    print("add tos doc id:", rsp.result.doc_id)


    url_uri = "https://your-url.pdf"
    rsp = collection.add_doc_v2(
        AddDocV2Request(
            doc_id="your-doc-id",
            doc_name="your-file-name.pdf",
            uri=url_uri,
        )
    )
    print("add url doc id:", rsp.result.doc_id)


    query = "your query"
    resp = collection.search_knowledge(
        SearchKnowledgeRequest(
            query=query,
            limit=10,
            dense_weight=0.5,
        )
    )


    top_items = []
    for item in resp.result.result_list[:5]:
        title = item.chunk_title
        content = item.content
        top_items.append(f"【{title}】\n{content}")


    context_text = "\n\n".join(top_items) if top_items else "（检索结果为空或不可用）"
    system_prompt = (
        "你是一位专业的财报分析师，"
        "你需要根据「参考资料」来回答接下来的「用户问题」，这些信息在 <context></context> XML 标签之内。"
        "回答必须在参考资料范围内，尽可能简洁，无法回答时请礼貌说明并引导提供更多信息。"
        "\n\n<context>\n" + context_text + "\n</context>"
    )
    msgs = [
        ChatMessage(role="system", content=system_prompt),
        ChatMessage(role="user", content=query),
    ]


    req = ChatCompletionRequest(
        model="Doubao-1-5-pro-32k",
        messages=msgs,
        thinking=None,
        max_tokens=4096,
        temperature=0.1,
        return_token_usage=True,
        stream=False,
    )
    res = client.chat_completion(req)
    print("chat_completion:", res.model_dump(by_alias=True))




if __name__ == "__main__":
    try:
        main()
    except VikingKnowledgeException as e:
        print(f"Exception Code: {e.code}")
        print(f"Exception Message: {e.message}")
        print(f"Exception Request ID: {e.request_id}")
        sys.exit(1)
    except Exception as e:
        print(e)
        sys.exit(1)

最近更新时间：2026.03.20 12:23:12

这个页面对您有帮助吗？

有用

无用