Not getting relavant results for AWS KENDRARetriver #26143
Replies: 13 comments 21 replies
-
|
I found a similar unsolved discussion that might be relevant to your issue:
Additionally, here are some potential reasons and troubleshooting steps for why the code might not be fetching relevant results from AWS Kendra, even though the same text search in the Kendra console returns a perfect match:
Here is the relevant code snippet for the def _kendra_query(self, query: str) -> Sequence[ResultItem]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.retrieve(**kendra_kwargs)
r_result = RetrieveResult.parse_obj(response)
if r_result.ResultItems:
return r_result.ResultItems
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return q_result.ResultItemsEnsure that the If all these factors are correct and the issue persists, further investigation into the specific implementation of the |
Beta Was this translation helpful? Give feedback.
-
|
i am getting results but it is not the same when compared with console |
Beta Was this translation helpful? Give feedback.
-
|
even i have passed specific data source id to fetch the info from very specific document @dosu. |
Beta Was this translation helpful? Give feedback.
-
|
The discrepancy in results between the AWS Kendra console and the In the provided code, if the def _kendra_query(self, query: str) -> Sequence[ResultItem]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.retrieve(**kendra_kwargs)
r_result = RetrieveResult.parse_obj(response)
if r_result.ResultItems:
return r_result.ResultItems
# Retrieve API returned 0 results, fall back to Query API
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return q_result.ResultItemsThis fallback mechanism means that if the |
Beta Was this translation helpful? Give feedback.
-
|
how can i fix this @dosu. |
Beta Was this translation helpful? Give feedback.
-
|
To fix the discrepancy in results between the AWS Kendra console and the
Here is an example of how to configure the from langchain_community.retrievers.kendra import AmazonKendraRetriever
retriever = AmazonKendraRetriever(
index_id="your_index_id",
region_name="your_region",
credentials_profile_name="your_profile_name",
top_k=3,
attribute_filter={
'EqualsTo': {
'Key': '_language_code',
'Value': {'StringValue': 'en'}
},
'EqualsTo': {
'Key': '_data_source_id',
'Value': {'StringValue': '1ce733e9d085'}
}
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
],
user_context={"your_user_context"}
)Ensure that the parameters in the |
Beta Was this translation helpful? Give feedback.
-
|
if you can help me with the complete code |
Beta Was this translation helpful? Give feedback.
-
|
To fix the discrepancy in results between the AWS Kendra console and the import re
from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Union
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document
from langchain_core.pydantic_v1 import BaseModel, Field, root_validator, validator
from langchain_core.retrievers import BaseRetriever
from typing_extensions import Annotated
def clean_excerpt(excerpt: str) -> str:
if not excerpt:
return excerpt
res = re.sub(r"\s+", " ", excerpt).replace("...", "")
return res
def combined_text(item: "ResultItem") -> str:
text = ""
title = item.get_title()
if title:
text += f"Document Title: {title}\n"
excerpt = clean_excerpt(item.get_excerpt())
if excerpt:
text += f"Document Excerpt: \n{excerpt}\n"
return text
DocumentAttributeValueType = Union[str, int, List[str], None]
class Highlight(BaseModel, extra="allow"):
BeginOffset: int
EndOffset: int
TopAnswer: Optional[bool]
Type: Optional[str]
class TextWithHighLights(BaseModel, extra="allow"):
Text: str
Highlights: Optional[Any]
class AdditionalResultAttributeValue(BaseModel, extra="allow"):
TextWithHighlightsValue: TextWithHighLights
class AdditionalResultAttribute(BaseModel, extra="allow"):
Key: str
ValueType: Literal["TEXT_WITH_HIGHLIGHTS_VALUE"]
Value: AdditionalResultAttributeValue
def get_value_text(self) -> str:
return self.Value.TextWithHighlightsValue.Text
class DocumentAttributeValue(BaseModel, extra="allow"):
DateValue: Optional[str]
LongValue: Optional[int]
StringListValue: Optional[List[str]]
StringValue: Optional[str]
@property
def value(self) -> DocumentAttributeValueType:
if self.DateValue:
return self.DateValue
if self.LongValue:
return self.LongValue
if self.StringListValue:
return self.StringListValue
if self.StringValue:
return self.StringValue
return None
class DocumentAttribute(BaseModel, extra="allow"):
Key: str
Value: DocumentAttributeValue
class ResultItem(BaseModel, ABC, extra="allow"):
Id: Optional[str]
DocumentId: Optional[str]
DocumentURI: Optional[str]
DocumentAttributes: Optional[List[DocumentAttribute]] = []
ScoreAttributes: Optional[dict]
@abstractmethod
def get_title(self) -> str:
pass
@abstractmethod
def get_excerpt(self) -> str:
pass
def get_additional_metadata(self) -> dict:
return {}
def get_document_attributes_dict(self) -> Dict[str, DocumentAttributeValueType]:
return {attr.Key: attr.Value.value for attr in (self.DocumentAttributes or [])}
def get_score_attribute(self) -> str:
if self.ScoreAttributes is not None:
return self.ScoreAttributes["ScoreConfidence"]
else:
return "NOT_AVAILABLE"
def to_doc(self, page_content_formatter: Callable[["ResultItem"], str] = combined_text) -> Document:
page_content = page_content_formatter(self)
metadata = self.get_additional_metadata()
metadata.update(
{
"result_id": self.Id,
"document_id": self.DocumentId,
"source": self.DocumentURI,
"title": self.get_title(),
"excerpt": self.get_excerpt(),
"document_attributes": self.get_document_attributes_dict(),
"score": self.get_score_attribute(),
}
)
return Document(page_content=page_content, metadata=metadata)
class QueryResultItem(ResultItem):
DocumentTitle: TextWithHighLights
FeedbackToken: Optional[str]
Format: Optional[str]
Type: Optional[str]
AdditionalAttributes: Optional[List[AdditionalResultAttribute]] = []
DocumentExcerpt: Optional[TextWithHighLights]
def get_title(self) -> str:
return self.DocumentTitle.Text
def get_attribute_value(self) -> str:
if not self.AdditionalAttributes:
return ""
if not self.AdditionalAttributes[0]:
return ""
else:
return self.AdditionalAttributes[0].get_value_text()
def get_excerpt(self) -> str:
if self.AdditionalAttributes and self.AdditionalAttributes[0].Key == "AnswerText":
excerpt = self.get_attribute_value()
elif self.DocumentExcerpt:
excerpt = self.DocumentExcerpt.Text
else:
excerpt = ""
return excerpt
def get_additional_metadata(self) -> dict:
additional_metadata = {"type": self.Type}
return additional_metadata
class RetrieveResultItem(ResultItem):
DocumentTitle: Optional[str]
Content: Optional[str]
def get_title(self) -> str:
return self.DocumentTitle or ""
def get_excerpt(self) -> str:
return self.Content or ""
class QueryResult(BaseModel, extra="allow"):
ResultItems: List[QueryResultItem]
class RetrieveResult(BaseModel, extra="allow"):
QueryId: str
ResultItems: List[RetrieveResultItem]
KENDRA_CONFIDENCE_MAPPING = {
"NOT_AVAILABLE": 0.0,
"LOW": 0.25,
"MEDIUM": 0.50,
"HIGH": 0.75,
"VERY_HIGH": 1.0,
}
class AmazonKendraRetriever(BaseRetriever):
index_id: str
region_name: Optional[str] = None
credentials_profile_name: Optional[str] = None
top_k: int = 3
attribute_filter: Optional[Dict] = None
document_relevance_override_configurations: Optional[List[Dict]] = None
page_content_formatter: Callable[[ResultItem], str] = combined_text
client: Any
user_context: Optional[Dict] = None
min_score_confidence: Annotated[Optional[float], Field(ge=0.0, le=1.0)]
@validator("top_k")
def validate_top_k(cls, value: int) -> int:
if value < 0:
raise ValueError(f"top_k ({value}) cannot be negative.")
return value
@root_validator(pre=True)
def create_client(cls, values: Dict[str, Any]) -> Dict[str, Any]:
if values.get("client") is not None:
return values
try:
import boto3
if values.get("credentials_profile_name"):
session = boto3.Session(profile_name=values["credentials_profile_name"])
else:
session = boto3.Session()
client_params = {}
if values.get("region_name"):
client_params["region_name"] = values["region_name"]
values["client"] = session.client("kendra", **client_params)
return values
except ImportError:
raise ImportError(
"Could not import boto3 python package. "
"Please install it with `pip install boto3`."
)
except Exception as e:
raise ValueError(
"Could not load credentials to authenticate with AWS client. "
"Please check that credentials in the specified "
"profile name are valid."
) from e
def _kendra_query(self, query: str) -> Sequence[ResultItem]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.retrieve(**kendra_kwargs)
r_result = RetrieveResult.parse_obj(response)
if r_result.ResultItems:
return r_result.ResultItems
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return q_result.ResultItems
def _get_top_k_docs(self, result_items: Sequence[ResultItem]) -> List[Document]:
top_docs = [
item.to_doc(self.page_content_formatter)
for item in result_items[: self.top_k]
]
return top_docs
def _filter_by_score_confidence(self, docs: List[Document]) -> List[Document]:
if not self.min_score_confidence:
return docs
filtered_docs = [
item
for item in docs
if (
item.metadata.get("score") is not None
and isinstance(item.metadata["score"], str)
and KENDRA_CONFIDENCE_MAPPING.get(item.metadata["score"], 0.0)
>= self.min_score_confidence
)
]
return filtered_docs
def _get_relevant_documents(
self,
query: str,
*,
run_manager: CallbackManagerForRetrieverRun,
) -> List[Document]:
result_items = self._kendra_query(query)
top_k_docs = self._get_top_k_docs(result_items)
return self._filter_by_score_confidence(top_k_docs)Make sure to set up your environment variables correctly: export AWS_DEFAULT_REGION=<your-region>
export AWS_PROFILE=<your-profile>
export KENDRA_INDEX_ID=<your-kendra-index-id>Additionally, ensure that Note: The |
Beta Was this translation helpful? Give feedback.
-
|
Since i am using aws lambda i dont need to setup the AWS config. |
Beta Was this translation helpful? Give feedback.
-
|
Here is the code for configuring the import boto3
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from typing import Optional, Dict, List, Callable, Any, Sequence
from langchain_core.pydantic_v1 import Field
from typing_extensions import Annotated
class AmazonKendraRetriever(BaseRetriever):
"""`Amazon Kendra Index` retriever.
Args:
index_id: Kendra index id
region_name: The aws region e.g., `us-west-2`.
credentials_profile_name: The name of the profile in the ~/.aws/credentials
or ~/.aws/config files, which has either access keys or role information
specified. If not specified, the default credential profile or, if on an
EC2 instance, credentials from IMDS will be used.
top_k: No of results to return
attribute_filter: Additional filtering of results based on metadata
document_relevance_override_configurations: Overrides relevance tuning
configurations of fields/attributes set at the index level
page_content_formatter: generates the Document page_content
allowing access to all result item attributes. By default, it uses
the item's title and excerpt.
client: boto3 client for Kendra
user_context: Provides information about the user context
"""
index_id: str
region_name: Optional[str] = None
credentials_profile_name: Optional[str] = None
top_k: int = 3
attribute_filter: Optional[Dict] = None
document_relevance_override_configurations: Optional[List[Dict]] = None
page_content_formatter: Callable[[Any], str] = lambda x: x.get("Text", "")
client: Any
user_context: Optional[Dict] = None
min_score_confidence: Annotated[Optional[float], Field(ge=0.0, le=1.0)]
@root_validator(pre=True)
def create_client(cls, values: Dict[str, Any]) -> Dict[str, Any]:
if values.get("client") is not None:
return values
try:
import boto3
if values.get("credentials_profile_name"):
session = boto3.Session(profile_name=values["credentials_profile_name"])
else:
session = boto3.Session()
client_params = {}
if values.get("region_name"):
client_params["region_name"] = values["region_name"]
values["client"] = session.client("kendra", **client_params)
return values
except ImportError:
raise ImportError(
"Could not import boto3 python package. "
"Please install it with `pip install boto3`."
)
except Exception as e:
raise ValueError(
"Could not load credentials to authenticate with AWS client. "
"Please check that credentials in the specified "
"profile name are valid."
) from e
def _kendra_query(self, query: str) -> Sequence[Any]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.retrieve(**kendra_kwargs)
r_result = response.get("ResultItems", [])
if r_result:
return r_result
response = self.client.query(**kendra_kwargs)
q_result = response.get("ResultItems", [])
return q_result
def _get_top_k_docs(self, result_items: Sequence[Any]) -> List[Document]:
top_docs = [
Document(page_content=self.page_content_formatter(item), metadata=item)
for item in result_items[: self.top_k]
]
return top_docs
def _filter_by_score_confidence(self, docs: List[Document]) -> List[Document]:
if not self.min_score_confidence:
return docs
filtered_docs = [
item
for item in docs
if (
item.metadata.get("Score") is not None
and isinstance(item.metadata["Score"], str)
and KENDRA_CONFIDENCE_MAPPING.get(item.metadata["Score"], 0.0)
>= self.min_score_confidence
)
]
return filtered_docs
def _get_relevant_documents(
self,
query: str,
*,
run_manager: Any,
) -> List[Document]:
result_items = self._kendra_query(query)
top_k_docs = self._get_top_k_docs(result_items)
return self._filter_by_score_confidence(top_k_docs)
# Example usage
retriever = AmazonKendraRetriever(index_id="your_kendra_index_id")
retriever.invoke("your_query")In an AWS Lambda environment, ensure that the Lambda function has the necessary IAM permissions to access Amazon Kendra and that the AWS SDK (boto3) is available. The AWS Lambda environment will automatically handle the AWS credentials, so you don't need to set up the AWS config manually [1][2]. |
Beta Was this translation helpful? Give feedback.
-
|
how can i Integrate above code with below code def build_chain(): #llm = OpenAI(batch_size=5, temperature=0, max_tokens=300) The base URL for your Azure OpenAI resource. e.g. "https://.openai.azure.com"#gpt 4 config API version e.g. "2023-07-01-preview"#openai_api_version="2024-03-01-preview" The name of your Azure OpenAI deployment chat model. e.g. "gpt-35-turbo-0613"#deployment_name="gpt-35-turbo" The API key for your Azure OpenAI resource.openai_api_key = os.environ["openai_api_key"] This is set to
|
Beta Was this translation helpful? Give feedback.
-
|
To integrate the provided AWS Lambda-compatible
This code sets up the |
Beta Was this translation helpful? Give feedback.
-
|
but this code does not have Kendraretriver custom class |
Beta Was this translation helpful? Give feedback.
-
|
To integrate the custom
This code sets up the |
Beta Was this translation helpful? Give feedback.
-
|
this sma ecode i am using but not getting the desired result |
Beta Was this translation helpful? Give feedback.
-
|
To address the issue of not getting the desired results with the
Here is the complete code with these considerations: import os
from langchain_community.retrievers import AmazonKendraRetriever
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain.chains.conversation.memory import ConversationBufferMemory
from prompts import prompt_template1
def build_chain():
region = os.environ["REGION"]
kendra_index_id = os.environ["KENDRA_INDEX"]
azure_endpoint = os.environ["azure_endpoint"]
openai_api_version = "01-preview"
deployment_name = "gpt-4o"
openai_api_key = os.environ["openai_api_key"]
openai_api_type = "azure"
llm = AzureChatOpenAI(
azure_endpoint=azure_endpoint,
openai_api_version=openai_api_version,
deployment_name=deployment_name,
openai_api_key=openai_api_key,
openai_api_type=openai_api_type,
)
data_source_id = '1ce733e9d085'
retriever = AmazonKendraRetriever(
index_id=kendra_index_id,
top_k=10, # Increase top_k to retrieve more documents
attribute_filter={
'EqualsTo': {
'Key': '_language_code',
'Value': {'StringValue': 'en'}
},
'EqualsTo': {
'Key': '_data_source_id',
'Value': {'StringValue': data_source_id}
}
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
]
)
prompt_template = prompt_template1
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
condense_qa_template = """
Given the following conversation and a follow up question, rephrase the follow up question
to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
standalone_question_prompt = PromptTemplate.from_template(condense_qa_template)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, input_key='question', output_key='answer')
qa = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_question_prompt=standalone_question_prompt,
return_source_documents=True,
memory=memory,
combine_docs_chain_kwargs={"prompt": PROMPT}
)
return qa
def run_chain(chain, prompt: str, history=[]):
print("History from run chain function", history)
return chain({"question": prompt, "chat_history": history})Key Points:
This setup should help in retrieving more relevant documents and improving the results [1][2]. |
Beta Was this translation helpful? Give feedback.
-
|
Kendra console and ConversationalRetrievalChain resullts are not matching. Using aws lambda to run the code so config is correct but. getting other not matching result from the same document. |
Beta Was this translation helpful? Give feedback.
-
|
The discrepancy between the results from the AWS Kendra console and the Here is the relevant code snippet: def _kendra_query(self, query: str) -> Sequence[ResultItem]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.retrieve(**kendra_kwargs)
r_result = RetrieveResult.parse_obj(response)
if r_result.ResultItems:
return r_result.ResultItems
# Retrieve API returned 0 results, fall back to Query API
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return q_result.ResultItemsIf the Steps to Address the Issue:
Here is the complete code with these considerations: import os
from langchain_community.retrievers import AmazonKendraRetriever
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain.chains.conversation.memory import ConversationBufferMemory
from prompts import prompt_template1
def build_chain():
region = os.environ["REGION"]
kendra_index_id = os.environ["KENDRA_INDEX"]
azure_endpoint = os.environ["azure_endpoint"]
openai_api_version = "01-preview"
deployment_name = "gpt-4o"
openai_api_key = os.environ["openai_api_key"]
openai_api_type = "azure"
llm = AzureChatOpenAI(
azure_endpoint=azure_endpoint,
openai_api_version=openai_api_version,
deployment_name=deployment_name,
openai_api_key=openai_api_key,
openai_api_type=openai_api_type,
)
data_source_id = '1ce733e9d085'
retriever = AmazonKendraRetriever(
index_id=kendra_index_id,
top_k=10, # Increase top_k to retrieve more documents
attribute_filter={
'EqualsTo': {
'Key': '_language_code',
'Value': {'StringValue': 'en'}
},
'EqualsTo': {
'Key': '_data_source_id',
'Value': {'StringValue': data_source_id}
}
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
]
)
prompt_template = prompt_template1
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
condense_qa_template = """
Given the following conversation and a follow up question, rephrase the follow up question
to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
standalone_question_prompt = PromptTemplate.from_template(condense_qa_template)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, input_key='question', output_key='answer')
qa = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_question_prompt=standalone_question_prompt,
return_source_documents=True,
memory=memory,
combine_docs_chain_kwargs={"prompt": PROMPT}
)
return qa
def run_chain(chain, prompt: str, history=[]):
print("History from run chain function", history)
return chain({"question": prompt, "chat_history": history})This setup should help in retrieving more relevant documents and improving the results [1][2]. |
Beta Was this translation helpful? Give feedback.
-
|
How can i check whether retrive query is returning zero results |
Beta Was this translation helpful? Give feedback.
-
|
To check whether the Here is the relevant code snippet: response = self.client.retrieve(**kendra_kwargs)
r_result = RetrieveResult.parse_obj(response)
if r_result.ResultItems:
return r_result.ResultItems
# Retrieve API returned 0 results, fall back to Query API
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return q_result.ResultItemsIn this code, the check |
Beta Was this translation helpful? Give feedback.
-
|
is it posiible to check this from here def run_chain(chain, prompt: str, history=[]): |
Beta Was this translation helpful? Give feedback.
-
|
Yes, it is possible to check if the retrieve query is returning zero results from within the Here is an example of how you can modify the from langchain_core.documents import Document
from langchain_core.language_models import FakeListLLM
from langchain.chains.conversational_retrieval.base import ConversationalRetrievalChain
from langchain.memory.buffer import ConversationBufferMemory
from tests.unit_tests.retrievers.sequential_retriever import SequentialRetriever
async def test_zero_results_handling() -> None:
fixed_resp = "No results found."
llm = FakeListLLM(responses=["I know the answer!"])
retriever = SequentialRetriever(sequential_responses=[[]])
memory = ConversationBufferMemory(
k=1, output_key="answer", memory_key="chat_history", return_messages=True
)
qa_chain = ConversationalRetrievalChain.from_llm(
llm=llm,
memory=memory,
retriever=retriever,
return_source_documents=True,
rephrase_question=False,
response_if_no_docs_found=fixed_resp,
verbose=True,
)
got = await qa_chain.acall("What is the answer?")
assert got["chat_history"][1].content == fixed_resp
assert got["answer"] == fixed_respIn this setup, the Additionally, you can use query analysis to decide whether to call the retriever or not based on the query analysis result. This can be useful in cases where the query analysis technique may allow for any number of queries to be generated, including no queries. Here is an example of how to handle such cases: from typing import Optional
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers.openai_tools import PydanticToolsParser
from langchain_core.runnables import chain
class Search(BaseModel):
query: str = Field(
...,
description="Similarity search query applied to job record.",
)
system = """You have the ability to issue search queries to get information to help answer user information.
You do not NEED to look things up. If you don't need to, then just respond normally."""
prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("human", "{question}"),
]
)
llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
structured_llm = llm.bind_tools([Search])
query_analyzer = {"question": RunnablePassthrough()} | prompt | structured_llm
output_parser = PydanticToolsParser(tools=[Search])
@chain
def custom_chain(question):
response = query_analyzer.invoke(question)
if "tool_calls" in response.additional_kwargs:
query = output_parser.invoke(response)
docs = retriever.invoke(query[0].query)
return docs
else:
return responseIn this example, the |
Beta Was this translation helpful? Give feedback.
-
|
When I am explicitly calling query api with filters getting the right result Can you please guide me if i can use the above code response in ConversationalRetrievalChain |
Beta Was this translation helpful? Give feedback.
-
|
Yes, you can use the response from the explicit Kendra query API call with filters in the Here is how you can integrate the explicit Kendra query API call with the
Here is the complete code: import os
import boto3
from typing import Sequence, Dict, Any
from langchain_community.retrievers import AmazonKendraRetriever
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain.chains.conversation.memory import ConversationBufferMemory
from prompts import prompt_template1
class CustomAmazonKendraRetriever(AmazonKendraRetriever):
def _kendra_query(self, query: str) -> Sequence[Dict[str, Any]]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.query(**kendra_kwargs)
q_result = response.get("ResultItems", [])
return q_result
def build_chain():
region = os.environ["REGION"]
kendra_index_id = os.environ["KENDRA_INDEX"]
azure_endpoint = os.environ["azure_endpoint"]
openai_api_version = "01-preview"
deployment_name = "gpt-4o"
openai_api_key = os.environ["openai_api_key"]
openai_api_type = "azure"
llm = AzureChatOpenAI(
azure_endpoint=azure_endpoint,
openai_api_version=openai_api_version,
deployment_name=deployment_name,
openai_api_key=openai_api_key,
openai_api_type=openai_api_type,
)
data_source_id = '1ce733e9d085'
retriever = CustomAmazonKendraRetriever(
index_id=kendra_index_id,
top_k=10, # Increase top_k to retrieve more documents
attribute_filter={
'AndAllFilters': [
{"EqualsTo": {"Key": "_language_code", "Value": {"StringValue": "en"}}},
{"EqualsTo": {"Key": "_data_source_id", "Value": {"StringValue": data_source_id}}}
]
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
]
)
prompt_template = prompt_template1
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
condense_qa_template = """
Given the following conversation and a follow up question, rephrase the follow up question
to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
standalone_question_prompt = PromptTemplate.from_template(condense_qa_template)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, input_key='question', output_key='answer')
qa = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_question_prompt=standalone_question_prompt,
return_source_documents=True,
memory=memory,
combine_docs_chain_kwargs={"prompt": PROMPT}
)
return qa
def run_chain(chain, prompt: str, history=[]):
print("History from run chain function", history)
return chain({"question": prompt, "chat_history": history})
# Example usage
if __name__ == "__main__":
chain = build_chain()
result = run_chain(chain, "Your query here")
print(result)Key Points:
This setup should help in retrieving more relevant documents and improving the results [1][2][3]. |
Beta Was this translation helpful? Give feedback.
-
|
[ERROR] AttributeError: 'dict' object has no attribute 'to_doc' |
Beta Was this translation helpful? Give feedback.
-
|
The error To fix this, you need to ensure that the items in
Here is the updated code: import os
import boto3
from typing import Sequence, Dict, Any, List
from langchain_community.retrievers import AmazonKendraRetriever
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain.chains.conversation.memory import ConversationBufferMemory
from prompts import prompt_template1
class CustomAmazonKendraRetriever(AmazonKendraRetriever):
def _kendra_query(self, query: str) -> Sequence[Dict[str, Any]]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return [QueryResultItem(**item) for item in q_result.ResultItems]
def build_chain():
region = os.environ["REGION"]
kendra_index_id = os.environ["KENDRA_INDEX"]
azure_endpoint = os.environ["azure_endpoint"]
openai_api_version = "01-preview"
deployment_name = "gpt-4o"
openai_api_key = os.environ["openai_api_key"]
openai_api_type = "azure"
llm = AzureChatOpenAI(
azure_endpoint=azure_endpoint,
openai_api_version=openai_api_version,
deployment_name=deployment_name,
openai_api_key=openai_api_key,
openai_api_type=openai_api_type,
)
data_source_id = '1ce733e9d085'
retriever = CustomAmazonKendraRetriever(
index_id=kendra_index_id,
top_k=10, # Increase top_k to retrieve more documents
attribute_filter={
'AndAllFilters': [
{"EqualsTo": {"Key": "_language_code", "Value": {"StringValue": "en"}}},
{"EqualsTo": {"Key": "_data_source_id", "Value": {"StringValue": data_source_id}}}
]
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
]
)
prompt_template = prompt_template1
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
condense_qa_template = """
Given the following conversation and a follow up question, rephrase the follow up question
to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
standalone_question_prompt = PromptTemplate.from_template(condense_qa_template)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, input_key='question', output_key='answer')
qa = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_question_prompt=standalone_question_prompt,
return_source_documents=True,
memory=memory,
combine_docs_chain_kwargs={"prompt": PROMPT}
)
return qa
def run_chain(chain, prompt: str, history=[]):
print("History from run chain function", history)
return chain({"question": prompt, "chat_history": history})
# Example usage
if __name__ == "__main__":
chain = build_chain()
result = run_chain(chain, "Your query here")
print(result)Key Points:
This setup should help in retrieving more relevant documents and improving the results [1]. |
Beta Was this translation helpful? Give feedback.
-
|
[ERROR] NameError: name 'QueryResult' is not defined |
Beta Was this translation helpful? Give feedback.
-
|
To resolve the Here is how you can import the from langchain_community.retrievers.kendra import QueryResult, QueryResultItemMake sure to add this import statement at the top of your script where you define the Here is the updated code with the necessary import: import os
import boto3
from typing import Sequence, Dict, Any, List
from langchain_community.retrievers.kendra import AmazonKendraRetriever, QueryResult, QueryResultItem
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain.chains.conversation.memory import ConversationBufferMemory
from prompts import prompt_template1
class CustomAmazonKendraRetriever(AmazonKendraRetriever):
def _kendra_query(self, query: str) -> Sequence[QueryResultItem]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return [QueryResultItem(**item) for item in q_result.ResultItems]
def build_chain():
region = os.environ["REGION"]
kendra_index_id = os.environ["KENDRA_INDEX"]
azure_endpoint = os.environ["azure_endpoint"]
openai_api_version = "01-preview"
deployment_name = "gpt-4o"
openai_api_key = os.environ["openai_api_key"]
openai_api_type = "azure"
llm = AzureChatOpenAI(
azure_endpoint=azure_endpoint,
openai_api_version=openai_api_version,
deployment_name=deployment_name,
openai_api_key=openai_api_key,
openai_api_type=openai_api_type,
)
data_source_id = '1ce733e9d085'
retriever = CustomAmazonKendraRetriever(
index_id=kendra_index_id,
top_k=10, # Increase top_k to retrieve more documents
attribute_filter={
'AndAllFilters': [
{"EqualsTo": {"Key": "_language_code", "Value": {"StringValue": "en"}}},
{"EqualsTo": {"Key": "_data_source_id", "Value": {"StringValue": data_source_id}}}
]
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
]
)
prompt_template = prompt_template1
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
condense_qa_template = """
Given the following conversation and a follow up question, rephrase the follow up question
to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
standalone_question_prompt = PromptTemplate.from_template(condense_qa_template)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, input_key='question', output_key='answer')
qa = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_question_prompt=standalone_question_prompt,
return_source_documents=True,
memory=memory,
combine_docs_chain_kwargs={"prompt": PROMPT}
)
return qa
def run_chain(chain, prompt: str, history=[]):
print("History from run chain function", history)
return chain({"question": prompt, "chat_history": history})
# Example usage
if __name__ == "__main__":
chain = build_chain()
result = run_chain(chain, "Your query here")
print(result)Key Points:
This setup should help in retrieving more relevant documents and improving the results [1]. |
Beta Was this translation helpful? Give feedback.
-
|
[ERROR] TypeError: langchain_community.retrievers.kendra.QueryResultItem() argument after ** must be a mapping, not QueryResultItem |
Beta Was this translation helpful? Give feedback.
-
|
To resolve the The error occurs because the Here is an example of how you can modify the import os
import boto3
from typing import Sequence, Dict, Any, List
from langchain_community.retrievers.kendra import AmazonKendraRetriever, QueryResult, QueryResultItem
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain.chains.conversation.memory import ConversationBufferMemory
from prompts import prompt_template1
class CustomAmazonKendraRetriever(AmazonKendraRetriever):
def _kendra_query(self, query: str) -> Sequence[Dict[str, Any]]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return [item.dict() for item in q_result.ResultItems]
def build_chain():
region = os.environ["REGION"]
kendra_index_id = os.environ["KENDRA_INDEX"]
azure_endpoint = os.environ["azure_endpoint"]
openai_api_version = "01-preview"
deployment_name = "gpt-4o"
openai_api_key = os.environ["openai_api_key"]
openai_api_type = "azure"
llm = AzureChatOpenAI(
azure_endpoint=azure_endpoint,
openai_api_version=openai_api_version,
deployment_name=deployment_name,
openai_api_key=openai_api_key,
openai_api_type=openai_api_type,
)
data_source_id = '1ce733e9d085'
retriever = CustomAmazonKendraRetriever(
index_id=kendra_index_id,
top_k=10, # Increase top_k to retrieve more documents
attribute_filter={
'AndAllFilters': [
{"EqualsTo": {"Key": "_language_code", "Value": {"StringValue": "en"}}},
{"EqualsTo": {"Key": "_data_source_id", "Value": {"StringValue": data_source_id}}}
]
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
]
)
prompt_template = prompt_template1
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
condense_qa_template = """
Given the following conversation and a follow up question, rephrase the follow up question
to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
standalone_question_prompt = PromptTemplate.from_template(condense_qa_template)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, input_key='question', output_key='answer')
qa = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_question_prompt=standalone_question_prompt,
return_source_documents=True,
memory=memory,
combine_docs_chain_kwargs={"prompt": PROMPT}
)
return qa
def run_chain(chain, prompt: str, history=[]):
print("History from run chain function", history)
return chain({"question": prompt, "chat_history": history})
# Example usage
if __name__ == "__main__":
chain = build_chain()
result = run_chain(chain, "Your query here")
print(result)Key Points:
This setup should help in retrieving more relevant documents and improving the results [1]. |
Beta Was this translation helpful? Give feedback.
-
|
again getting to_doc error |
Beta Was this translation helpful? Give feedback.
-
|
The Here is the updated code to ensure proper conversion:
Here is the updated code: import os
import boto3
from typing import Sequence, Dict, Any, List
from langchain_community.retrievers.kendra import AmazonKendraRetriever, QueryResult, QueryResultItem
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain.chains.conversation.memory import ConversationBufferMemory
from prompts import prompt_template1
class CustomAmazonKendraRetriever(AmazonKendraRetriever):
def _kendra_query(self, query: str) -> Sequence[QueryResultItem]:
kendra_kwargs = {
"IndexId": self.index_id,
"QueryText": query.strip()[0:999],
"PageSize": self.top_k,
}
if self.attribute_filter is not None:
kendra_kwargs["AttributeFilter"] = self.attribute_filter
if self.document_relevance_override_configurations is not None:
kendra_kwargs["DocumentRelevanceOverrideConfigurations"] = (
self.document_relevance_override_configurations
)
if self.user_context is not None:
kendra_kwargs["UserContext"] = self.user_context
response = self.client.query(**kendra_kwargs)
q_result = QueryResult.parse_obj(response)
return [QueryResultItem(**item.dict()) for item in q_result.ResultItems]
def build_chain():
region = os.environ["REGION"]
kendra_index_id = os.environ["KENDRA_INDEX"]
azure_endpoint = os.environ["azure_endpoint"]
openai_api_version = "01-preview"
deployment_name = "gpt-4o"
openai_api_key = os.environ["openai_api_key"]
openai_api_type = "azure"
llm = AzureChatOpenAI(
azure_endpoint=azure_endpoint,
openai_api_version=openai_api_version,
deployment_name=deployment_name,
openai_api_key=openai_api_key,
openai_api_type=openai_api_type,
)
data_source_id = '1ce733e9d085'
retriever = CustomAmazonKendraRetriever(
index_id=kendra_index_id,
top_k=10, # Increase top_k to retrieve more documents
attribute_filter={
'AndAllFilters': [
{"EqualsTo": {"Key": "_language_code", "Value": {"StringValue": "en"}}},
{"EqualsTo": {"Key": "_data_source_id", "Value": {"StringValue": data_source_id}}}
]
},
document_relevance_override_configurations=[
{
"Name": "OverrideForDataSource",
"Relevance": {
"Freshness": True,
"Importance": 9,
"RankOrder": "ASCENDING"
}
}
]
)
prompt_template = prompt_template1
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
condense_qa_template = """
Given the following conversation and a follow up question, rephrase the follow up question
to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
standalone_question_prompt = PromptTemplate.from_template(condense_qa_template)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, input_key='question', output_key='answer')
qa = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_question_prompt=standalone_question_prompt,
return_source_documents=True,
memory=memory,
combine_docs_chain_kwargs={"prompt": PROMPT}
)
return qa
def run_chain(chain, prompt: str, history=[]):
print("History from run chain function", history)
return chain({"question": prompt, "chat_history": history})
# Example usage
if __name__ == "__main__":
chain = build_chain()
result = run_chain(chain, "Your query here")
print(result)Key Points:
This setup should help in retrieving more relevant documents and improving the results [1]. |
Beta Was this translation helpful? Give feedback.
-
|
getting same erors again and again.. this code will get dict error. Doing same thing again and again |
Beta Was this translation helpful? Give feedback.
-
|
Thanks this solved the issue . now its fetching the expected results |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
-
Checked other resources
Commit to Help
Example Code
Description
I am using above code to implement the que ans based chatbot. but i am facing few issues in search. when I am searching for any specific text through above code I am not getting relevant result . but if i search the same text in Kendra console getting the perfect match for the resonse.
So my question is even there is perfect match is available in Kendra and from console i am able to get the answer .why the above code is not fetching the same result
System Info
NA
Beta Was this translation helpful? Give feedback.
All reactions