# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
import functools
from typing import Any, Callable, Dict, Generic, Optional, TypeVar, overload, Union, List
import warnings
from azure.core.exceptions import (
ClientAuthenticationError,
HttpResponseError,
ResourceExistsError,
ResourceNotFoundError,
map_error,
)
from azure.core.pipeline import PipelineResponse
from azure.core.pipeline.transport import AsyncHttpResponse
from azure.core.rest import HttpRequest
from ... import models as _models
from ...operations._operations import build_query_knowledge_base_request, build_query_text_request
from ..._patch import _validate_text_records, _get_positional_body, _verify_qna_id_and_question
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]]
[docs]class QuestionAnsweringClientOperationsMixin:
@overload
async def query_knowledge_base(
self,
options: "_models.QueryKnowledgeBaseOptions",
*,
project_name: str,
deployment_name: str,
**kwargs: Any
) -> "_models.KnowledgeBaseAnswers":
"""Answers the specified question using your knowledge base.
:param options: Positional-only POST body of the request.
:type options:
~azure.ai.language.questionanswering.models.QueryKnowledgeBaseOptions
:keyword project_name: The name of the project to use.
:paramtype project_name: str
:keyword deployment_name: The name of the specific deployment of the project to use.
:paramtype deployment_name: str
:return: KnowledgeBaseAnswers
:rtype: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswers
:raises: ~azure.core.exceptions.HttpResponseError
"""
...
@overload
async def query_knowledge_base(
self,
*,
project_name: str,
deployment_name: str,
qna_id: Optional[int] = None,
question: Optional[str] = None,
top: Optional[int] = None,
user_id: Optional[str] = None,
confidence_score_threshold: Optional[float] = None,
context: Optional["_models.KnowledgeBaseAnswerRequestContext"] = None,
ranker_type: Optional[Union[str, "_models.RankerType"]] = None,
filters: Optional["_models.QueryFilters"] = None,
answer_span_request: Optional["_models.AnswerSpanRequest"] = None,
include_unstructured_sources: Optional[bool] = None,
**kwargs: Any
) -> "_models.KnowledgeBaseAnswers":
"""Answers the specified question using your knowledge base.
:keyword project_name: The name of the project to use.
:paramtype project_name: str
:keyword deployment_name: The name of the specific deployment of the project to use.
:paramtype deployment_name: str
:keyword qna_id: Exact QnA ID to fetch from the knowledge base, this field takes priority over
question.
:paramtype qna_id: int
:keyword question: User question to query against the knowledge base.
:paramtype question: str
:keyword top: Max number of answers to be returned for the question.
:paramtype top: int
:keyword user_id: Unique identifier for the user.
:paramtype user_id: str
:keyword confidence_score_threshold: Minimum threshold score for answers, value ranges from 0 to
1.
:paramtype confidence_score_threshold: float
:keyword context: Context object with previous QnA's information.
:paramtype context: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswerRequestContext
:keyword ranker_type: (Optional) Set to 'QuestionOnly' for using a question only Ranker. Possible
values include: "Default", "QuestionOnly".
:paramtype ranker_type: str or ~azure.ai.language.questionanswering.models.RankerType
:keyword filters: Filter QnAs based on give metadata list and knowledge base source names.
:paramtype filters: ~azure.ai.language.questionanswering.models.QueryFilters
:keyword answer_span_request: To configure Answer span prediction feature.
:paramtype answer_span_request: ~azure.ai.language.questionanswering.models.AnswerSpanRequest
:keyword include_unstructured_sources: (Optional) Flag to enable Query over Unstructured Sources.
:paramtype include_unstructured_sources: bool
:rtype: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswers
:raises: ~azure.core.exceptions.HttpResponseError
"""
...
[docs] async def query_knowledge_base(
self,
*args,
**kwargs: Any
) -> "_models.KnowledgeBaseAnswers":
"""Answers the specified question using your knowledge base.
:param options: POST body of the request. Provide either `options`, OR
individual keyword arguments. If both are provided, only the options object will be used.
:type options:
~azure.ai.language.questionanswering.models.QueryKnowledgeBaseOptions
:keyword project_name: The name of the project to use.
:paramtype project_name: str
:keyword deployment_name: The name of the specific deployment of the project to use.
:paramtype deployment_name: str
:keyword qna_id: Exact QnA ID to fetch from the knowledge base, this field takes priority over question.
:paramtype qna_id: int
:keyword question: User question to query against the knowledge base. Provide either `options`, OR
individual keyword arguments. If both are provided, only the options object will be used.
:paramtype question: str
:keyword top: Max number of answers to be returned for the question.
:paramtype top: int
:keyword user_id: Unique identifier for the user.
:paramtype user_id: str
:keyword confidence_score_threshold: Minimum threshold score for answers, value ranges from 0 to 1.
:paramtype confidence_score_threshold: float
:keyword context: Context object with previous QnA's information.
:paramtype context: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswerRequestContext
:keyword ranker_type: (Optional) Set to 'QuestionOnly' for using a question only Ranker. Possible
values include: "Default", "QuestionOnly".
:paramtype ranker_type: str or ~azure.ai.language.questionanswering.models.RankerType
:keyword filters: Filter QnAs based on give metadata list and knowledge base source names.
:paramtype filters: ~azure.ai.language.questionanswering.models.QueryFilters
:keyword answer_span_request: To configure Answer span prediction feature.
:paramtype answer_span_request: ~azure.ai.language.questionanswering.models.AnswerSpanRequest
:keyword include_unstructured_sources: (Optional) Flag to enable Query over Unstructured Sources.
:paramtype include_unstructured_sources: bool
:return: KnowledgeBaseAnswers
:rtype: ~azure.ai.language.questionanswering.models.KnowledgeBaseAnswers
:raises: ~azure.core.exceptions.HttpResponseError
"""
options = _get_positional_body(*args, **kwargs) or _models.QueryKnowledgeBaseOptions(
qna_id=kwargs.pop("qna_id", None),
question=kwargs.pop("question", None),
top=kwargs.pop("top", None),
user_id=kwargs.pop("user_id", None),
confidence_score_threshold=kwargs.pop("confidence_score_threshold", None),
context=kwargs.pop("context", None),
ranker_type=kwargs.pop("ranker_type", None),
filters=kwargs.pop("filters", None),
answer_span_request=kwargs.pop("answer_span_request", None),
include_unstructured_sources=kwargs.pop("include_unstructured_sources", None)
)
_verify_qna_id_and_question(options)
cls = kwargs.pop("cls", None) # type: ClsType["_models.KnowledgeBaseAnswers"]
error_map = {401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError}
error_map.update(kwargs.pop("error_map", {}))
content_type = kwargs.pop("content_type", "application/json") # type: Optional[str]
project_name = kwargs.pop("project_name") # type: str
deployment_name = kwargs.pop("deployment_name") # type: str
json = self._serialize.body(options, "QueryKnowledgeBaseOptions")
request = build_query_knowledge_base_request(
content_type=content_type,
project_name=project_name,
deployment_name=deployment_name,
json=json,
template_url=self.query_knowledge_base.metadata["url"],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str", skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
raise HttpResponseError(response=response, model=error)
deserialized = self._deserialize("KnowledgeBaseAnswers", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
query_knowledge_base.metadata = {"url": "/:query-knowledgebases"} # type: ignore
@overload
async def query_text(self, options: "_models.QueryTextOptions", **kwargs: Any) -> "_models.TextAnswers":
"""Answers the specified question using the provided text in the body.
:param options: Positional-only POST body of the request.
:type options: ~azure.ai.language.questionanswering.models.QueryTextOptions
:return: TextAnswers
:rtype: ~azure.ai.language.questionanswering.models.TextAnswers
:raises: ~azure.core.exceptions.HttpResponseError
"""
...
@overload
async def query_text(
self,
*,
question: str,
records: List["_models.TextRecord"],
language: Optional[str] = None,
**kwargs: Any
) -> "_models.TextAnswers":
"""Answers the specified question using the provided text in the body.
:keyword question: Required. User question to query against the given text records.
:paramtype question: str
:keyword records: Required. Text records to be searched for given question.
:paramtype records: list[~azure.ai.language.questionanswering.models.TextRecord]
:keyword language: Language of the text records. This is BCP-47 representation of a language. For
example, use "en" for English; "es" for Spanish etc. If not set, use "en" for English as
default.
:paramtype language: str
:return: TextAnswers
:rtype: ~azure.ai.language.questionanswering.models.TextAnswers
:raises: ~azure.core.exceptions.HttpResponseError
"""
...
[docs] async def query_text(
self, *args, **kwargs: Any
) -> "_models.TextAnswers":
"""Answers the specified question using the provided text in the body.
:param options: POST body of the request. Provide either `options`, OR
individual keyword arguments. If both are provided, only the options object will be used.
:type options: ~azure.ai.language.questionanswering.models.QueryTextOptions
:keyword question: User question to query against the given text records. Provide either `options`, OR
individual keyword arguments. If both are provided, only the options object will be used.
:paramtype question: str
:keyword records: Text records to be searched for given question. Provide either `options`, OR
individual keyword arguments. If both are provided, only the options object will be used.
:paramtype records: list[~azure.ai.language.questionanswering.models.TextRecord]
:keyword language: Language of the text records. This is BCP-47 representation of a language. For
example, use "en" for English; "es" for Spanish etc. If not set, use "en" for English as default.
:paramtype language: str
:return: TextAnswers
:rtype: ~azure.ai.language.questionanswering.models.TextAnswers
:raises: ~azure.core.exceptions.HttpResponseError
"""
options = _get_positional_body(*args, **kwargs) or _models.QueryTextOptions(
question=kwargs.pop("question"),
records=kwargs.pop("records"),
language=kwargs.pop("language", self._default_language),
)
try:
options['records'] = _validate_text_records(options['records'])
except TypeError:
options.records = _validate_text_records(options.records)
cls = kwargs.pop("cls", None) # type: ClsType["_models.TextAnswers"]
error_map = {401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError}
error_map.update(kwargs.pop("error_map", {}))
content_type = kwargs.pop("content_type", "application/json") # type: Optional[str]
json = self._serialize.body(options, "QueryTextOptions")
request = build_query_text_request(
content_type=content_type,
json=json,
template_url=self.query_text.metadata["url"],
)
path_format_arguments = {
"Endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str", skip_quote=True),
}
request.url = self._client.format_url(request.url, **path_format_arguments)
pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
raise HttpResponseError(response=response, model=error)
deserialized = self._deserialize("TextAnswers", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {})
return deserialized
query_text.metadata = {"url": "/:query-text"} # type: ignore