File: sample_query_knowledgebase_async.py

package info (click to toggle)
python-azure 20251118%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 783,356 kB
  • sloc: python: 6,474,533; ansic: 804; javascript: 287; sh: 205; makefile: 198; xml: 109
file content (64 lines) | stat: -rw-r--r-- 2,028 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
# coding=utf-8
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------

"""
FILE: sample_query_knowledgebase_async.py

DESCRIPTION:
    Async knowledge base query using flattened parameters.

USAGE:
    python sample_query_knowledgebase_async.py
"""

from __future__ import annotations
import asyncio


async def sample_query_knowledgebase():
    # [START query_knowledgebase_async]
    import os
    from azure.core.credentials import AzureKeyCredential
    from azure.ai.language.questionanswering.aio import QuestionAnsweringClient
    from azure.ai.language.questionanswering.models import (
        AnswersOptions,
        ShortAnswerOptions,
    )

    endpoint = os.environ["AZURE_QUESTIONANSWERING_ENDPOINT"]
    key = os.environ["AZURE_QUESTIONANSWERING_KEY"]
    project = os.environ["AZURE_QUESTIONANSWERING_PROJECT"]
    deployment = os.environ.get("AZURE_QUESTIONANSWERING_DEPLOYMENT", "production")

    client = QuestionAnsweringClient(endpoint, AzureKeyCredential(key))
    async with client:
        question = "How long should my Surface battery last?"
        options = AnswersOptions(
            question=question,
            top=3,
            confidence_threshold=0.2,
            include_unstructured_sources=True,
            short_answer_options=ShortAnswerOptions(enable=True, confidence_threshold=0.2, top=1),
        )
        output = await client.get_answers(
            options,
            project_name=project,
            deployment_name=deployment,
        )
        best_candidate = next(
            (a for a in (output.answers or []) if a.confidence and a.confidence > 0.7),
            None,
        )
        if best_candidate:
            print(f"Q: {question}")
            print(f"A: {best_candidate.answer}")
        else:
            print(f"No answers for '{question}'")
    # [END query_knowledgebase_async]


if __name__ == "__main__":
    asyncio.run(sample_query_knowledgebase())