#152

pending completion

Build # #152

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

<a href="https://github.com/microsoft/botbuilder-python/commit/<a class=hub.com/microsoft/botbuilder-python/commit/6c2991cc7576edcd4a4844283d6a2f1a75ac4b1f">6c2991cc7<a href="https://github.com/microsoft/botbuilder-python/commit/6c2991cc7576edcd4a4844283d6a2f1a75ac4b1f">">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/5ed1c1bc69fecfb1f16b0494d6e43390fd321d87">5ed1c1bc6</a>">5ed1c1bc6</a><a href="https://github.com/microsoft/botbuilder-python/commit/6c2991cc7576edcd4a4844283d6a2f1a75ac4b1f"> into 6340a92fe">6340a92fe</a>

Run Details

42 of 42 new or added lines in 9 files covered. (100.0%)

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

92.86

/libraries/botbuilder-ai/botbuilder/ai/qna/qnamaker.py

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

import json
from typing import Dict, List, NamedTuple, Union
from aiohttp import ClientSession, ClientTimeout

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

from .models import FeedbackRecord, QueryResult
from .utils import (
    ActiveLearningUtils,
    GenerateAnswerUtils,
    QnATelemetryConstants,
    TrainUtils,
)
from .qnamaker_endpoint import QnAMakerEndpoint
from .qnamaker_options import QnAMakerOptions
from .qnamaker_telemetry_client import QnAMakerTelemetryClient

from .. import __title__, __version__


class EventData(NamedTuple):
    properties: Dict[str, str]
    metrics: Dict[str, float]


class QnAMaker(QnAMakerTelemetryClient):
    """
    Class used to query a QnA Maker knowledge base for answers.
    """

    def __init__(
        self,
        endpoint: QnAMakerEndpoint,
        options: QnAMakerOptions = None,
        http_client: ClientSession = None,
        telemetry_client: BotTelemetryClient = None,
        log_personal_information: bool = None,
    ):
        super().__init__(log_personal_information, telemetry_client)

        if not isinstance(endpoint, QnAMakerEndpoint):
            raise TypeError(
                "QnAMaker.__init__(): endpoint is not an instance of QnAMakerEndpoint"
            )

        self._endpoint: str = endpoint

        opt = options or QnAMakerOptions()
        self._validate_options(opt)

        instance_timeout = ClientTimeout(total=opt.timeout / 1000)
        self._http_client = http_client or ClientSession(timeout=instance_timeout)

        self.telemetry_client: Union[
            BotTelemetryClient, NullTelemetryClient
        ] = telemetry_client or NullTelemetryClient()

        self.log_personal_information = log_personal_information or False

        self._generate_answer_helper = GenerateAnswerUtils(
            self.telemetry_client, self._endpoint, options, self._http_client
        )
        self._active_learning_train_helper = TrainUtils(
            self._endpoint, self._http_client
        )

    async def get_answers(
        self,
        context: TurnContext,
        options: QnAMakerOptions = None,
        telemetry_properties: Dict[str, str] = None,
        telemetry_metrics: Dict[str, int] = None,
    ) -> [QueryResult]:
        """
        Generates answers from the knowledge base.

        return:
        -------
        A list of answers for the user's query, sorted in decreasing order of ranking score.

        rtype:
        ------
        List[QueryResult]
        """
        if not context:
            raise TypeError("QnAMaker.get_answers(): context cannot be None.")

        if not isinstance(context.activity, Activity):
            raise TypeError(
                "QnAMaker.get_answers(): TurnContext's activity must be an Activity instance."
            )

        result = await self._generate_answer_helper.get_answers(context, options)

        await self.on_qna_result(
            result, context, telemetry_properties, telemetry_metrics
        )

        return result

    def get_low_score_variation(self, query_result: QueryResult) -> List[QueryResult]:
        """
        Filters the ambiguous question for active learning.

        Parameters:
        -----------
        query_result: User query output.

        Return:
        -------
        Filtered aray of ambigous questions.
        """
        return ActiveLearningUtils.get_low_score_variation(query_result)

    async def call_train(self, feedback_records: List[FeedbackRecord]):
        """
        Sends feedback to the knowledge base.

        Parameters:
        -----------
        feedback_records
        """
        return await self._active_learning_train_helper.call_train(feedback_records)

    async def on_qna_result(
        self,
        query_results: [QueryResult],
        turn_context: TurnContext,
        telemetry_properties: Dict[str, str] = None,
        telemetry_metrics: Dict[str, float] = None,
    ):
        event_data = await self.fill_qna_event(
            query_results, turn_context, telemetry_properties, telemetry_metrics
        )

        # Track the event
        self.telemetry_client.track_event(
            name=QnATelemetryConstants.qna_message_event,
            properties=event_data.properties,
            measurements=event_data.metrics,
        )

    async def fill_qna_event(
        self,
        query_results: [QueryResult],
        turn_context: TurnContext,
        telemetry_properties: Dict[str, str] = None,
        telemetry_metrics: Dict[str, float] = None,
    ) -> EventData:
        """
        Fills the event properties and metrics for the QnaMessage event for telemetry.

        return:
        -------
        A tuple of event data properties and metrics that will be sent to the
        BotTelemetryClient.track_event() method for the QnAMessage event.
        The properties and metrics returned the standard properties logged
        with any properties passed from the get_answers() method.

        rtype:
        ------
        EventData
        """

        properties: Dict[str, str] = dict()
        metrics: Dict[str, float] = dict()

        properties[
            QnATelemetryConstants.knowledge_base_id_property
        ] = self._endpoint.knowledge_base_id

        text: str = turn_context.activity.text
        user_name: str = turn_context.activity.from_property.name

        # Use the LogPersonalInformation flag to toggle logging PII data; text and username are common examples.
        if self.log_personal_information:
            if text:
                properties[QnATelemetryConstants.question_property] = text

            if user_name:
                properties[QnATelemetryConstants.username_property] = user_name

        # Fill in Qna Results (found or not).
        if self._has_matched_answer_in_kb(query_results):
            query_result = query_results[0]

            result_properties = {
                QnATelemetryConstants.matched_question_property: json.dumps(
                    query_result.questions
                ),
                QnATelemetryConstants.question_id_property: str(query_result.id),
                QnATelemetryConstants.answer_property: query_result.answer,
                QnATelemetryConstants.article_found_property: "true",
            }
            properties.update(result_properties)

            metrics[QnATelemetryConstants.score_metric] = query_result.score
        else:
            no_match_properties = {
                QnATelemetryConstants.matched_question_property: "No Qna Question matched",
                QnATelemetryConstants.question_id_property: "No Qna Question Id matched",
                QnATelemetryConstants.answer_property: "No Qna Answer matched",
                QnATelemetryConstants.article_found_property: "false",
            }

            properties.update(no_match_properties)

        # Additional Properties can override "stock" properties.
        if telemetry_properties:
            properties.update(telemetry_properties)

        # Additional Metrics can override "stock" metrics.
        if telemetry_metrics:
            metrics.update(telemetry_metrics)

        return EventData(properties=properties, metrics=metrics)

    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 _has_matched_answer_in_kb(self, query_results: [QueryResult]) -> bool:
        if query_results:
            if query_results[0].id != -1:

                return True

        return False

1	# Copyright (c) Microsoft Corporation. All rights reserved.
2	# Licensed under the MIT License.
3
4	import json	1✔
5	from typing import Dict, List, NamedTuple, Union	1✔
6	from aiohttp import ClientSession, ClientTimeout	1✔
7
8	from botbuilder.schema import Activity	1✔
9	from botbuilder.core import BotTelemetryClient, NullTelemetryClient, TurnContext	1✔
10
11	from .models import FeedbackRecord, QueryResult	1✔
12	from .utils import (	1✔
13	ActiveLearningUtils,
14	GenerateAnswerUtils,
15	QnATelemetryConstants,
16	TrainUtils,
17	)
18	from .qnamaker_endpoint import QnAMakerEndpoint	1✔
19	from .qnamaker_options import QnAMakerOptions	1✔
20	from .qnamaker_telemetry_client import QnAMakerTelemetryClient	1✔
21
22	from .. import __title__, __version__	1✔
23
24
25	class EventData(NamedTuple):	1✔
26	properties: Dict[str, str]	1✔
27	metrics: Dict[str, float]	1✔
28
29
30	class QnAMaker(QnAMakerTelemetryClient):	1✔
31	"""
32	Class used to query a QnA Maker knowledge base for answers.
33	"""
34
35	def __init__(	1✔
36	self,
37	endpoint: QnAMakerEndpoint,
38	options: QnAMakerOptions = None,
39	http_client: ClientSession = None,
40	telemetry_client: BotTelemetryClient = None,
41	log_personal_information: bool = None,
42	):
43	super().__init__(log_personal_information, telemetry_client)	1✔
44
45	if not isinstance(endpoint, QnAMakerEndpoint):	1✔
46	raise TypeError(	1✔
47	"QnAMaker.__init__(): endpoint is not an instance of QnAMakerEndpoint"
48	)
49
50	self._endpoint: str = endpoint	1✔
51
52	opt = options or QnAMakerOptions()	1✔
53	self._validate_options(opt)	1✔
54
55	instance_timeout = ClientTimeout(total=opt.timeout / 1000)	1✔
56	self._http_client = http_client or ClientSession(timeout=instance_timeout)	1✔
57
58	self.telemetry_client: Union[	1✔
59	BotTelemetryClient, NullTelemetryClient
60	] = telemetry_client or NullTelemetryClient()
61
62	self.log_personal_information = log_personal_information or False	1✔
63
64	self._generate_answer_helper = GenerateAnswerUtils(	1✔
65	self.telemetry_client, self._endpoint, options, self._http_client
66	)
67	self._active_learning_train_helper = TrainUtils(	1✔
68	self._endpoint, self._http_client
69	)
70
71	async def get_answers(	1✔
72	self,
73	context: TurnContext,
74	options: QnAMakerOptions = None,
75	telemetry_properties: Dict[str, str] = None,
76	telemetry_metrics: Dict[str, int] = None,
77	) -> [QueryResult]:
78	"""
79	Generates answers from the knowledge base.
80
81	return:
82	-------
83	A list of answers for the user's query, sorted in decreasing order of ranking score.
84
85	rtype:
86	------
87	List[QueryResult]
88	"""
89	if not context:	1✔
90	raise TypeError("QnAMaker.get_answers(): context cannot be None.")	×
91
92	if not isinstance(context.activity, Activity):	1✔
93	raise TypeError(	×
94	"QnAMaker.get_answers(): TurnContext's activity must be an Activity instance."
95	)
96
97	result = await self._generate_answer_helper.get_answers(context, options)	1✔
98
99	await self.on_qna_result(	1✔
100	result, context, telemetry_properties, telemetry_metrics
101	)
102
103	return result	1✔
104
105	def get_low_score_variation(self, query_result: QueryResult) -> List[QueryResult]:	1✔
106	"""
107	Filters the ambiguous question for active learning.
108
109	Parameters:
110	-----------
111	query_result: User query output.
112
113	Return:
114	-------
115	Filtered aray of ambigous questions.
116	"""
117	return ActiveLearningUtils.get_low_score_variation(query_result)	1✔
118
119	async def call_train(self, feedback_records: List[FeedbackRecord]):	1✔
120	"""
121	Sends feedback to the knowledge base.
122
123	Parameters:
124	-----------
125	feedback_records
126	"""
127	return await self._active_learning_train_helper.call_train(feedback_records)	×
128
129	async def on_qna_result(	1✔
130	self,
131	query_results: [QueryResult],
132	turn_context: TurnContext,
133	telemetry_properties: Dict[str, str] = None,
134	telemetry_metrics: Dict[str, float] = None,
135	):
136	event_data = await self.fill_qna_event(	1✔
137	query_results, turn_context, telemetry_properties, telemetry_metrics
138	)
139
140	# Track the event
141	self.telemetry_client.track_event(	1✔
142	name=QnATelemetryConstants.qna_message_event,
143	properties=event_data.properties,
144	measurements=event_data.metrics,
145	)
146
147	async def fill_qna_event(	1✔
148	self,
149	query_results: [QueryResult],
150	turn_context: TurnContext,
151	telemetry_properties: Dict[str, str] = None,
152	telemetry_metrics: Dict[str, float] = None,
153	) -> EventData:
154	"""
155	Fills the event properties and metrics for the QnaMessage event for telemetry.
156
157	return:
158	-------
159	A tuple of event data properties and metrics that will be sent to the
160	BotTelemetryClient.track_event() method for the QnAMessage event.
161	The properties and metrics returned the standard properties logged
162	with any properties passed from the get_answers() method.
163
164	rtype:
165	------
166	EventData
167	"""
168
169	properties: Dict[str, str] = dict()	1✔
170	metrics: Dict[str, float] = dict()	1✔
171
172	properties[	1✔
173	QnATelemetryConstants.knowledge_base_id_property
174	] = self._endpoint.knowledge_base_id
175
176	text: str = turn_context.activity.text	1✔
177	user_name: str = turn_context.activity.from_property.name	1✔
178
179	# Use the LogPersonalInformation flag to toggle logging PII data; text and username are common examples.
180	if self.log_personal_information:	1✔
181	if text:	1✔
182	properties[QnATelemetryConstants.question_property] = text	1✔
183
184	if user_name:	1✔
185	properties[QnATelemetryConstants.username_property] = user_name	×
186
187	# Fill in Qna Results (found or not).
188	if self._has_matched_answer_in_kb(query_results):	1✔
189	query_result = query_results[0]	1✔
190
191	result_properties = {	1✔
192	QnATelemetryConstants.matched_question_property: json.dumps(
193	query_result.questions
194	),
195	QnATelemetryConstants.question_id_property: str(query_result.id),
196	QnATelemetryConstants.answer_property: query_result.answer,
197	QnATelemetryConstants.article_found_property: "true",
198	}
199	properties.update(result_properties)	1✔
200
201	metrics[QnATelemetryConstants.score_metric] = query_result.score	1✔
202	else:
203	no_match_properties = {	1✔
204	QnATelemetryConstants.matched_question_property: "No Qna Question matched",
205	QnATelemetryConstants.question_id_property: "No Qna Question Id matched",
206	QnATelemetryConstants.answer_property: "No Qna Answer matched",
207	QnATelemetryConstants.article_found_property: "false",
208	}
209
210	properties.update(no_match_properties)	1✔
211
212	# Additional Properties can override "stock" properties.
213	if telemetry_properties:	1✔
214	properties.update(telemetry_properties)	1✔
215
216	# Additional Metrics can override "stock" metrics.
217	if telemetry_metrics:	1✔
218	metrics.update(telemetry_metrics)	1✔
219
220	return EventData(properties=properties, metrics=metrics)	1✔
221
222	def _validate_options(self, options: QnAMakerOptions):	1✔
223	if not options.score_threshold:	1✔
224	options.score_threshold = 0.3	1✔
225
226	if not options.top:	1✔
227	options.top = 1	1✔
228
229	if options.score_threshold < 0 or options.score_threshold > 1:	1✔
230	raise ValueError("Score threshold should be a value between 0 and 1")	×
231
232	if options.top < 1:	1✔
233	raise ValueError("QnAMakerOptions.top should be an integer greater than 0")	×
234
235	if not options.strict_filters:	1✔
236	options.strict_filters = []	1✔
237
238	if not options.timeout:	1✔
239	options.timeout = 100000	1✔
240
241	def _has_matched_answer_in_kb(self, query_results: [QueryResult]) -> bool:	1✔
242	if query_results:	1✔
243	if query_results[0].id != -1:	1✔
244
245	return True	1✔
246
247	return False	1✔

microsoft / botbuilder-python / #152

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