#172

pending completion

Build # #172

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/df106751ea41dffeaf5649bf7ec4588f04f1e5c0">df106751e<a href="https://github.com/microsoft/botbuilder-python/commit/df106751ea41dffeaf5649bf7ec4588f04f1e5c0">">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/b533313507fb43321ddc4f158ad387eff2318d6c">b53331350</a>">b53331350</a><a href="https://github.com/microsoft/botbuilder-python/commit/df106751ea41dffeaf5649bf7ec4588f04f1e5c0"> into 63ebf2272">63ebf2272</a>

Run Details

5 of 5 new or added lines in 4 files covered. (100.0%)

4204 of 5573 relevant lines covered (75.44%)

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

            hydrated_options.context = query_options.context
            hydrated_options.qna_id = query_options.qna_id

        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,
            qna_id=options.qna_id,
        )

        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,
            qna_id=options.qna_id,
        )

        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	hydrated_options.context = query_options.context	1✔
131	hydrated_options.qna_id = query_options.qna_id	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	qna_id=options.qna_id,
147	)
148
149	http_request_helper = HttpRequestUtils(self._http_client)	1✔
150
151	response: ClientResponse = await http_request_helper.execute_http_request(	1✔
152	url, question, self._endpoint, options.timeout
153	)
154
155	result: List[QueryResult] = await self._format_qna_result(response, options)	1✔
156
157	return result	1✔
158
159	async def _emit_trace_info(	1✔
160	self, context: TurnContext, result: List[QueryResult], options: QnAMakerOptions
161	):
162	trace_info = QnAMakerTraceInfo(	1✔
163	message=context.activity,
164	query_results=result,
165	knowledge_base_id=self._endpoint.knowledge_base_id,
166	score_threshold=options.score_threshold,
167	top=options.top,
168	strict_filters=options.strict_filters,
169	context=options.context,
170	qna_id=options.qna_id,
171	)
172
173	trace_activity = Activity(	1✔
174	label=QNAMAKER_TRACE_LABEL,
175	name=QNAMAKER_TRACE_NAME,
176	type="trace",
177	value=trace_info,
178	value_type=QNAMAKER_TRACE_TYPE,
179	)
180
181	await context.send_activity(trace_activity)	1✔
182
183	async def _format_qna_result(	1✔
184	self, result, options: QnAMakerOptions
185	) -> List[QueryResult]:
186	json_res = result	1✔
187	if isinstance(result, ClientResponse):	1✔
188	json_res = await result.json()	×
189
190	answers_within_threshold = [	1✔
191	{**answer, "score": answer["score"] / 100}
192	for answer in json_res["answers"]
193	if answer["score"] / 100 > options.score_threshold
194	]
195	sorted_answers = sorted(	1✔
196	answers_within_threshold, key=lambda ans: ans["score"], reverse=True
197	)
198
199	answers_as_query_results = list(	1✔
200	map(lambda answer: QueryResult(**answer), sorted_answers)
201	)
202
203	return answers_as_query_results	1✔

microsoft / botbuilder-python / #172

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