#167

pending completion

Build # #167

Build Type

push

python-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/microsoft/botbuilder-python/commit/<a class=hub.com/microsoft/botbuilder-python/commit/41ab61700042a6e3d2f8076f436ab2023c00e78e">41ab61700<a href="https://github.com/microsoft/botbuilder-python/commit/41ab61700042a6e3d2f8076f436ab2023c00e78e">">Merge </a><a class="double-link" href="https://github.com/microsoft/botbuilder-python/commit/<a class="double-link" href="https://github.com/microsoft/botbuilder-python/commit/6e7ecddc77e3be80aefbd9ee309efc8d53a0635a">6e7ecddc7</a>">6e7ecddc7</a><a href="https://github.com/microsoft/botbuilder-python/commit/41ab61700042a6e3d2f8076f436ab2023c00e78e"> into 31dd7acde">31dd7acde</a>

Run Details

4201 of 5570 relevant lines covered (75.42%)

0.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.15

/libraries/botbuilder-ai/botbuilder/ai/qna/utils/generate_answer_utils.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

from copy import copy
from typing import List, Union

from aiohttp import ClientResponse, ClientSession

from botbuilder.core import BotTelemetryClient, NullTelemetryClient, TurnContext
from botbuilder.schema import Activity

from .http_request_utils import HttpRequestUtils

from ..qnamaker_endpoint import QnAMakerEndpoint
from ..qnamaker_options import QnAMakerOptions
from ..models import GenerateAnswerRequestBody, QnAMakerTraceInfo, QueryResult

QNAMAKER_TRACE_NAME = "QnAMaker"
QNAMAKER_TRACE_LABEL = "QnAMaker Trace"
QNAMAKER_TRACE_TYPE = "https://www.qnamaker.ai/schemas/trace"


class GenerateAnswerUtils:
    """
    Helper class for Generate Answer API, which is used to make queries to
    a single QnA Maker knowledge base and return the result.
    """

    def __init__(
        self,
        telemetry_client: Union[BotTelemetryClient, NullTelemetryClient],
        endpoint: QnAMakerEndpoint,
        options: QnAMakerOptions,
        http_client: ClientSession,
    ):
        """
        Parameters:
        -----------

        telemetry_client: Telemetry client.

        endpoint: QnA Maker endpoint details.

        options: QnA Maker options to configure the instance.

        http_client: HTTP client.
        """
        self._telemetry_client = telemetry_client
        self._endpoint = endpoint

        self.options = (
            options if isinstance(options, QnAMakerOptions) else QnAMakerOptions()
        )
        self._validate_options(self.options)

        self._http_client = http_client

    async def get_answers(
        self, context: TurnContext, options: QnAMakerOptions = None
    ) -> List[QueryResult]:
        if not isinstance(context, TurnContext):
            raise TypeError(
                "GenerateAnswerUtils.get_answers(): context must be an instance of TurnContext"
            )

        hydrated_options = self._hydrate_options(options)
        self._validate_options(hydrated_options)

        result: List[QueryResult] = await self._query_qna_service(
            context, hydrated_options
        )

        await self._emit_trace_info(context, result, hydrated_options)

        return result

    def _validate_options(self, options: QnAMakerOptions):
        if not options.score_threshold:
            options.score_threshold = 0.3

        if not options.top:
            options.top = 1

        if options.score_threshold < 0 or options.score_threshold > 1:
            raise ValueError("Score threshold should be a value between 0 and 1")

        if options.top < 1:
            raise ValueError("QnAMakerOptions.top should be an integer greater than 0")

        if not options.strict_filters:
            options.strict_filters = []

        if not options.timeout:
            options.timeout = 100000

    def _hydrate_options(self, query_options: QnAMakerOptions) -> QnAMakerOptions:
        """
        Combines QnAMakerOptions passed into the QnAMaker constructor
        with the options passed as arguments into get_answers().
        Return:
        -------
        QnAMakerOptions with options passed into constructor overwritten by new options passed into get_answers()

        rtype:
        ------
        QnAMakerOptions
        """

        hydrated_options = copy(self.options)

        if query_options:
            if (
                query_options.score_threshold != hydrated_options.score_threshold
                and query_options.score_threshold
            ):
                hydrated_options.score_threshold = query_options.score_threshold

            if query_options.top != hydrated_options.top and query_options.top != 0:
                hydrated_options.top = query_options.top

            if query_options.strict_filters:
                hydrated_options.strict_filters = query_options.strict_filters

            if (
                query_options.timeout != hydrated_options.timeout
                and query_options.timeout
            ):
                hydrated_options.timeout = query_options.timeout

            if query_options.context:
                hydrated_options.context = query_options.context

        return hydrated_options

    async def _query_qna_service(
        self, turn_context: TurnContext, options: QnAMakerOptions
    ) -> List[QueryResult]:
        url = f"{ self._endpoint.host }/knowledgebases/{ self._endpoint.knowledge_base_id }/generateAnswer"

        question = GenerateAnswerRequestBody(
            question=turn_context.activity.text,
            top=options.top,
            score_threshold=options.score_threshold,
            strict_filters=options.strict_filters,
            context=options.context,
        )

        http_request_helper = HttpRequestUtils(self._http_client)

        response: ClientResponse = await http_request_helper.execute_http_request(
            url, question, self._endpoint, options.timeout
        )

        result: List[QueryResult] = await self._format_qna_result(response, options)

        return result

    async def _emit_trace_info(
        self, context: TurnContext, result: List[QueryResult], options: QnAMakerOptions
    ):
        trace_info = QnAMakerTraceInfo(
            message=context.activity,
            query_results=result,
            knowledge_base_id=self._endpoint.knowledge_base_id,
            score_threshold=options.score_threshold,
            top=options.top,
            strict_filters=options.strict_filters,
            context=options.context,
        )

        trace_activity = Activity(
            label=QNAMAKER_TRACE_LABEL,
            name=QNAMAKER_TRACE_NAME,
            type="trace",
            value=trace_info,
            value_type=QNAMAKER_TRACE_TYPE,
        )

        await context.send_activity(trace_activity)

    async def _format_qna_result(
        self, result, options: QnAMakerOptions
    ) -> List[QueryResult]:
        json_res = result
        if isinstance(result, ClientResponse):
            json_res = await result.json()

        answers_within_threshold = [
            {**answer, "score": answer["score"] / 100}
            for answer in json_res["answers"]
            if answer["score"] / 100 > options.score_threshold
        ]
        sorted_answers = sorted(
            answers_within_threshold, key=lambda ans: ans["score"], reverse=True
        )

        answers_as_query_results = list(
            map(lambda answer: QueryResult(**answer), sorted_answers)
        )

        return answers_as_query_results

1	# Copyright (c) Microsoft Corporation. All rights reserved.
2	# Licensed under the MIT License.
3
4	from copy import copy	1✔
5	from typing import List, Union	1✔
6
7	from aiohttp import ClientResponse, ClientSession	1✔
8
9	from botbuilder.core import BotTelemetryClient, NullTelemetryClient, TurnContext	1✔
10	from botbuilder.schema import Activity	1✔
11
12	from .http_request_utils import HttpRequestUtils	1✔
13
14	from ..qnamaker_endpoint import QnAMakerEndpoint	1✔
15	from ..qnamaker_options import QnAMakerOptions	1✔
16	from ..models import GenerateAnswerRequestBody, QnAMakerTraceInfo, QueryResult	1✔
17
18	QNAMAKER_TRACE_NAME = "QnAMaker"	1✔
19	QNAMAKER_TRACE_LABEL = "QnAMaker Trace"	1✔
20	QNAMAKER_TRACE_TYPE = "https://www.qnamaker.ai/schemas/trace"	1✔
21
22
23	class GenerateAnswerUtils:	1✔
24	"""
25	Helper class for Generate Answer API, which is used to make queries to
26	a single QnA Maker knowledge base and return the result.
27	"""
28
29	def __init__(	1✔
30	self,
31	telemetry_client: Union[BotTelemetryClient, NullTelemetryClient],
32	endpoint: QnAMakerEndpoint,
33	options: QnAMakerOptions,
34	http_client: ClientSession,
35	):
36	"""
37	Parameters:
38	-----------
39
40	telemetry_client: Telemetry client.
41
42	endpoint: QnA Maker endpoint details.
43
44	options: QnA Maker options to configure the instance.
45
46	http_client: HTTP client.
47	"""
48	self._telemetry_client = telemetry_client	1✔
49	self._endpoint = endpoint	1✔
50
51	self.options = (	1✔
52	options if isinstance(options, QnAMakerOptions) else QnAMakerOptions()
53	)
54	self._validate_options(self.options)	1✔
55
56	self._http_client = http_client	1✔
57
58	async def get_answers(	1✔
59	self, context: TurnContext, options: QnAMakerOptions = None
60	) -> List[QueryResult]:
61	if not isinstance(context, TurnContext):	1✔
62	raise TypeError(	×
63	"GenerateAnswerUtils.get_answers(): context must be an instance of TurnContext"
64	)
65
66	hydrated_options = self._hydrate_options(options)	1✔
67	self._validate_options(hydrated_options)	1✔
68
69	result: List[QueryResult] = await self._query_qna_service(	1✔
70	context, hydrated_options
71	)
72
73	await self._emit_trace_info(context, result, hydrated_options)	1✔
74
75	return result	1✔
76
77	def _validate_options(self, options: QnAMakerOptions):	1✔
78	if not options.score_threshold:	1✔
79	options.score_threshold = 0.3	1✔
80
81	if not options.top:	1✔
82	options.top = 1	1✔
83
84	if options.score_threshold < 0 or options.score_threshold > 1:	1✔
85	raise ValueError("Score threshold should be a value between 0 and 1")	×
86
87	if options.top < 1:	1✔
88	raise ValueError("QnAMakerOptions.top should be an integer greater than 0")	×
89
90	if not options.strict_filters:	1✔
91	options.strict_filters = []	1✔
92
93	if not options.timeout:	1✔
94	options.timeout = 100000	1✔
95
96	def _hydrate_options(self, query_options: QnAMakerOptions) -> QnAMakerOptions:	1✔
97	"""
98	Combines QnAMakerOptions passed into the QnAMaker constructor
99	with the options passed as arguments into get_answers().
100	Return:
101	-------
102	QnAMakerOptions with options passed into constructor overwritten by new options passed into get_answers()
103
104	rtype:
105	------
106	QnAMakerOptions
107	"""
108
109	hydrated_options = copy(self.options)	1✔
110
111	if query_options:	1✔
112	if (	1✔
113	query_options.score_threshold != hydrated_options.score_threshold
114	and query_options.score_threshold
115	):
116	hydrated_options.score_threshold = query_options.score_threshold	1✔
117
118	if query_options.top != hydrated_options.top and query_options.top != 0:	1✔
119	hydrated_options.top = query_options.top	1✔
120
121	if query_options.strict_filters:	1✔
122	hydrated_options.strict_filters = query_options.strict_filters	1✔
123
124	if (	1✔
125	query_options.timeout != hydrated_options.timeout
126	and query_options.timeout
127	):
128	hydrated_options.timeout = query_options.timeout	×
129
130	if query_options.context:	1✔
131	hydrated_options.context = query_options.context	1✔
132
133	return hydrated_options	1✔
134
135	async def _query_qna_service(	1✔
136	self, turn_context: TurnContext, options: QnAMakerOptions
137	) -> List[QueryResult]:
138	url = f"{ self._endpoint.host }/knowledgebases/{ self._endpoint.knowledge_base_id }/generateAnswer"	1✔
139
140	question = GenerateAnswerRequestBody(	1✔
141	question=turn_context.activity.text,
142	top=options.top,
143	score_threshold=options.score_threshold,
144	strict_filters=options.strict_filters,
145	context=options.context,
146	)
147
148	http_request_helper = HttpRequestUtils(self._http_client)	1✔
149
150	response: ClientResponse = await http_request_helper.execute_http_request(	1✔
151	url, question, self._endpoint, options.timeout
152	)
153
154	result: List[QueryResult] = await self._format_qna_result(response, options)	1✔
155
156	return result	1✔
157
158	async def _emit_trace_info(	1✔
159	self, context: TurnContext, result: List[QueryResult], options: QnAMakerOptions
160	):
161	trace_info = QnAMakerTraceInfo(	1✔
162	message=context.activity,
163	query_results=result,
164	knowledge_base_id=self._endpoint.knowledge_base_id,
165	score_threshold=options.score_threshold,
166	top=options.top,
167	strict_filters=options.strict_filters,
168	context=options.context,
169	)
170
171	trace_activity = Activity(	1✔
172	label=QNAMAKER_TRACE_LABEL,
173	name=QNAMAKER_TRACE_NAME,
174	type="trace",
175	value=trace_info,
176	value_type=QNAMAKER_TRACE_TYPE,
177	)
178
179	await context.send_activity(trace_activity)	1✔
180
181	async def _format_qna_result(	1✔
182	self, result, options: QnAMakerOptions
183	) -> List[QueryResult]:
184	json_res = result	1✔
185	if isinstance(result, ClientResponse):	1✔
186	json_res = await result.json()	×
187
188	answers_within_threshold = [	1✔
189	{**answer, "score": answer["score"] / 100}
190	for answer in json_res["answers"]
191	if answer["score"] / 100 > options.score_threshold
192	]
193	sorted_answers = sorted(	1✔
194	answers_within_threshold, key=lambda ans: ans["score"], reverse=True
195	)
196
197	answers_as_query_results = list(	1✔
198	map(lambda answer: QueryResult(**answer), sorted_answers)
199	)
200
201	return answers_as_query_results	1✔

microsoft / botbuilder-python / #167

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous