Replies: 10 comments 118 replies
-
|
Hey @thisadev! I'm here to help you with any issues or questions you have. I'm a bot and can assist with bug fixes, answer your questions, and guide you on becoming a contributor. Let's take a look at the issue you're facing with the structured output. To use a Pydantic class called
Here's an example: from typing import List
from pydantic import BaseModel
from llama_index.legacy.llms.openai import OpenAI
from llama_index.legacy.prompts.base import PromptTemplate
from llama_index.legacy.output_parsers.pydantic import PydanticOutputParser
from llama_index.legacy.program.llm_program import LLMTextCompletionProgram
from llama_index.legacy.core.query_engine.retriever_query_engine import RetrieverQueryEngine
from llama_index.legacy.core.base.base_retriever import BaseRetriever
from llama_index.legacy.schema import QueryBundle
class Page(BaseModel):
"""Data model for a page."""
page: int
summary: str
# Define the prompt template
prompt_template_str = "Tell me about the history of Python programming language."
# Create the LLM instance (replace with your Azure Hosted LLM)
llm = OpenAI(temperature=0, model="gpt-3.5-turbo-0613")
# Create the output parser
output_parser = PydanticOutputParser(output_cls=Page)
# Create the prompt template
prompt = PromptTemplate(prompt_template_str)
# Create the LLMTextCompletionProgram
program = LLMTextCompletionProgram(
output_parser=output_parser,
output_cls=Page,
prompt=prompt,
llm=llm,
verbose=True
)
# Create a retriever (assuming you have a retriever implementation)
retriever = BaseRetriever()
# Create the RetrieverQueryEngine with the output_cls set to Page
query_engine = RetrieverQueryEngine.from_args(
retriever=retriever,
llm=llm,
output_cls=Page,
node_postprocessors=[lim_reorder, reranker],
response_synthesizer=response_synthesizer
)
# Query the engine and get the response
response = query_engine._query(query_bundle=QueryBundle(query_str=prompt_template_str))
print(response.page)
# > 1
print(response.summary)
# > 'Python was created by Guido van Rossum and first released in 1991...'This setup ensures that the |
Beta Was this translation helpful? Give feedback.
-
|
To ensure that your custom LLM works correctly with the
from pydantic import BaseModel
class Page(BaseModel):
"""Data model for a page."""
page: int
summary: str
Your class AzureHostedLLM(CustomLLM):
# ... (other methods and properties)
@llm_completion_callback()
def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
self.allow_self_signed_https(True)
data = {
"messages": [
{"role": "user", "content": prompt}
],
"max_tokens": self.num_output,
"temperature": 0.7,
"top_p": 1
}
headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)}
response_text = self.post_to_azure_api(self.api_url, data, headers)
# Ensure the response is a JSON string that matches the Page schema
response_json = json.loads(response_text)
return CompletionResponse(text=json.dumps(response_json))
from llama_index.legacy.core.base_retriever import BaseRetriever
from llama_index.legacy.query_engine.retriever_query_engine import RetrieverQueryEngine
from llama_index.legacy.callbacks.base import CallbackManager
from llama_index.legacy.response_synthesizers import get_response_synthesizer
# Assuming you have an instance of BaseRetriever
retriever = BaseRetriever()
# Create an instance of RetrieverQueryEngine with the Page class as output_cls
query_engine = RetrieverQueryEngine.from_args(
retriever=retriever,
output_cls=Page,
callback_manager=CallbackManager([]),
response_synthesizer=get_response_synthesizer(service_context=None)
)
query = f"""As an expert in evaluating company annual reports, use only the {company} Annual Report to identify the top 5 pages that best address the following check criteria.
Provide a brief summary (not more than 40 words.) for each page. Focus on the main body of the PDF, prioritizing the Sections listed below.
Exclude resulting any annexures, appendices, glossary, or index sections typically found at the end of the report.
Check criteria details: {criterion}
Answer in JSON format:
[{{
"page": [page_number],
"summary": "[brief_summary]"
}}]
"""
response = query_engine.query(query)
# Assuming response is a list of Page objects
for page in response:
print(page.page)
print(page.summary)By ensuring that the response from your custom LLM is correctly formatted as a JSON string that matches the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu Thank you for the reply. I followed the same approach and got the below error.I called the output_cls inside the response synthesizer.
Also, what I receive from the query response is something like below output.
|
Beta Was this translation helpful? Give feedback.
-
|
The error you're encountering indicates that the response from your custom LLM does not match the expected schema defined by the To resolve this, ensure that the response from your custom LLM is correctly formatted to match the
from pydantic import BaseModel
class Page(BaseModel):
"""Data model for a page."""
page: int
summary: str
Ensure that the response from the Azure API is parsed and formatted as a list of dictionaries that match the class AzureHostedLLM(CustomLLM):
# ... (other methods and properties)
@llm_completion_callback()
def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
self.allow_self_signed_https(True)
data = {
"messages": [
{"role": "user", "content": prompt}
],
"max_tokens": self.num_output,
"temperature": 0.7,
"top_p": 1
}
headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)}
response_text = self.post_to_azure_api(self.api_url, data, headers)
# Parse the response text to ensure it matches the Page schema
response_json = json.loads(response_text)
# Assuming the response is a list of pages
pages = []
for item in response_json:
page = Page(page=item['page'], summary=item['summary'])
pages.append(page)
return CompletionResponse(text=json.dumps([page.dict() for page in pages]))
from llama_index.legacy.core.base_retriever import BaseRetriever
from llama_index.legacy.query_engine.retriever_query_engine import RetrieverQueryEngine
from llama_index.legacy.callbacks.base import CallbackManager
from llama_index.legacy.response_synthesizers import get_response_synthesizer
# Assuming you have an instance of BaseRetriever
retriever = BaseRetriever()
# Create an instance of RetrieverQueryEngine with the Page class as output_cls
query_engine = RetrieverQueryEngine.from_args(
retriever=retriever,
output_cls=Page,
callback_manager=CallbackManager([]),
response_synthesizer=get_response_synthesizer(service_context=None)
)
query = f"""As an expert in evaluating company annual reports, use only the {company} Annual Report to identify the top 5 pages that best address the following check criteria.
Provide a brief summary (not more than 40 words.) for each page. Focus on the main body of the PDF, prioritizing the Sections listed below.
Exclude resulting any annexures, appendices, glossary, or index sections typically found at the end of the report.
Check criteria details: {criterion}
Answer in JSON format:
[{{
"page": [page_number],
"summary": "[brief_summary]"
}}]
"""
response = query_engine.query(query)
# Assuming response is a list of Page objects
for page in response:
print(page.page)
print(page.summary)By ensuring that the response from your custom LLM is correctly formatted as a JSON string that matches the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu After the changes in the completion callback of the LLM, the response output looks like the below. Using that format can you suggest me how to get the correct format from the CustomLLM to match the pydantic fields page and summary.
|
Beta Was this translation helpful? Give feedback.
-
|
To ensure that the response from your custom LLM matches the
from pydantic import BaseModel
class Page(BaseModel):
"""Data model for a page."""
page: int
summary: str
import json
from typing import Any, List
class AzureHostedLLM(CustomLLM):
# ... (other methods and properties)
@llm_completion_callback()
def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
self.allow_self_signed_https(True)
data = {
"messages": [
{"role": "user", "content": prompt}
],
"max_tokens": self.num_output,
"temperature": 0.7,
"top_p": 1
}
headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)}
response_text = self.post_to_azure_api(self.api_url, data, headers)
# Parse the response text
response_json = json.loads(response_text)
# Extract the content from the response
content = response_json["choices"][0]["message"]["content"]
# Parse the content to a list of Page objects
pages = json.loads(content)
page_objects = [Page(page=page["page"], summary=page["summary"]) for page in pages]
# Return the formatted response
return CompletionResponse(text=json.dumps([page.dict() for page in page_objects]))
# Example usage
query = f"""As an expert in evaluating company annual reports, use only the {company} Annual Report to identify the top 5 pages that best address the following check criteria.
Provide a brief summary (not more than 40 words.) for each page. Focus on the main body of the PDF, prioritizing the Sections listed below.
Exclude resulting any annexures, appendices, glossary, or index sections typically found at the end of the report.
Check criteria details: {criterion}
Answer in JSON format:
[{{
"page": [page_number],
"summary": "[brief_summary]"
}}]
"""
response = query_engine.query(query)
# Assuming response is a list of Page objects
for page in response:
print(page.page)
print(page.summary)This modification ensures that the response from your custom LLM is parsed and formatted to match the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I've tried an approach like below with my CustomLLM using the as_structured_llm. However I still get the pydantic validation error Below is the customLLM `import json class PageResults(BaseModel): class AzureHostedLLM(CustomLLM): ` And the below is how I called the Pydantic class as my output_cls in the response synthesizer. `prompt_helper = PromptHelper(context_window=4096, num_output=320, chunk_overlap_ratio=.2, chunk_size_limit=2048) |
Beta Was this translation helpful? Give feedback.
-
|
To use the
Ensure you have the def deserialize_streaming_output(response: bytes) -> str:
response_str = (
response.decode("utf-8").lstrip('[{"generated_text":"').rstrip('"}]')
)
clean_response = '{"response":"' + response_str + '"}'
return json.loads(clean_response)["response"]
Update your import json
import re
import logging
from typing import Any
from pydantic import BaseModel
from llama_index.core.llms import CustomLLM, CompletionResponse
from llama_index.core.llms.callbacks import llm_completion_callback
class PageResults(BaseModel):
"""Data model for page number and summary"""
page: int
summary: str
class AzureHostedLLM(CustomLLM):
context_window: int = 4096
num_output: int = 320
model_name: str = "azure_custom_model"
api_url: str = "api_url"
api_key: str = "api_key"
@property
def metadata(self) -> LLMMetadata:
"""Get LLM metadata."""
return LLMMetadata(
context_window=self.context_window,
num_output=self.num_output,
model_name=self.model_name
)
@staticmethod
def allow_self_signed_https(allowed):
if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None):
ssl._create_default_https_context = ssl._create_unverified_context
@staticmethod
def post_to_azure_api(url: str, data: dict, headers: dict):
body = json.dumps(data).encode('utf-8')
req = urllib.request.Request(url, body, headers)
try:
response = urllib.request.urlopen(req)
result = response.read().decode('utf-8')
return result
except urllib.error.HTTPError as error:
print("The request failed with status code: " + str(error.code))
print(error.info())
print(error.read().decode("utf-8"))
def deserialize_streaming_output(self, response: bytes) -> str:
response_str = (
response.decode("utf-8").lstrip('[{"generated_text":"').rstrip('"}]')
)
clean_response = '{"response":"' + response_str + '"}'
return json.loads(clean_response)["response"]
@llm_completion_callback()
def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
self.allow_self_signed_https(True)
data = {
"messages": [
{"role": "user", "content": f"Generate output in plain JSON format without any markdown, code block formatting, or additional commentary. Only return the JSON data. {prompt}"}
],
"max_tokens": self.num_output,
"temperature": 0.7,
"top_p": 1
}
headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)}
response_text = self.post_to_azure_api(self.api_url, data, headers)
# Log the raw JSON response
logging.debug(f"Raw JSON response: {response_text}")
# Strip unwanted formatting and extra data
response_text = response_text.replace("```json", "").replace("```", "").strip()
response_text = re.sub(r'[\x00-\x1F\x7F]', '', response_text)
try:
# Deserialize the response using the provided method
clean_response = self.deserialize_streaming_output(response_text.encode('utf-8'))
# Parse the cleaned response
response_json = json.loads(clean_response)
# Extract the content from the response
content = response_json["choices"][0]["message"]["content"]
# Strip unwanted formatting and extra data from the content
content = content.replace("```json", "").replace("```", "").strip()
content = re.sub(r'[\x00-\x1F\x7F]', '', content)
# Parse the content to a list of PageResults objects
pages = json.loads(content)
page_objects = []
for page in pages:
# Ensure page is always an integer
page_number = page["page"]
if isinstance(page_number, list):
page_number = page_number[0] if page_number else 0
page_objects.append(PageResults(page=int(page_number), summary=page["summary"]))
# Return the formatted response
return CompletionResponse(text=json.dumps([page.dict() for page in page_objects]))
except json.JSONDecodeError as e:
logging.error(f"JSON decoding error: {e}")
logging.error(f"Response text: {response_text}")
raise eBy integrating the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu What could be the reason for the below error? How can I clean this from JSON format?
|
Beta Was this translation helpful? Give feedback.
-
|
The error To clean and properly format the JSON response, you can use a function to handle incomplete or malformed JSON strings. Here’s how you can adjust your
You can use a function to handle incomplete or malformed JSON strings. import json
import re
import logging
from typing import Any
from pydantic import BaseModel
from llama_index.core.llms import CustomLLM, CompletionResponse
from llama_index.core.llms.callbacks import llm_completion_callback
class PageResults(BaseModel):
"""Data model for page number and summary"""
page: int
summary: str
class AzureHostedLLM(CustomLLM):
context_window: int = 4096
num_output: int = 320
model_name: str = "azure_custom_model"
api_url: str = "api_url"
api_key: str = "api_key"
@property
def metadata(self) -> LLMMetadata:
"""Get LLM metadata."""
return LLMMetadata(
context_window=self.context_window,
num_output=self.num_output,
model_name=self.model_name
)
@staticmethod
def allow_self_signed_https(allowed):
if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None):
ssl._create_default_https_context = ssl._create_unverified_context
@staticmethod
def post_to_azure_api(url: str, data: dict, headers: dict):
body = json.dumps(data).encode('utf-8')
req = urllib.request.Request(url, body, headers)
try:
response = urllib.request.urlopen(req)
result = response.read().decode('utf-8')
return result
except urllib.error.HTTPError as error:
print("The request failed with status code: " + str(error.code))
print(error.info())
print(error.read().decode("utf-8"))
def parse_partial_json(self, s: str) -> Any:
"""Parse an incomplete JSON string into a valid python dictionary."""
try:
return json.loads(s)
except json.JSONDecodeError:
pass
new_s = ""
stack = []
is_inside_string = False
escaped = False
for char in s:
if is_inside_string:
if char == '"' and not escaped:
is_inside_string = False
elif char == "\n" and not escaped:
char = "\\n"
elif char == "\\":
escaped = not escaped
else:
escaped = False
else:
if char == '"':
is_inside_string = True
escaped = False
elif char == "{":
stack.append("}")
elif char == "[":
stack.append("]")
elif char == "}" or char == "]":
if stack and stack[-1] == char:
stack.pop()
else:
raise ValueError("Malformed partial JSON encountered.")
new_s += char
if is_inside_string:
new_s += '"'
for closing_char in reversed(stack):
new_s += closing_char
try:
return json.loads(new_s)
except json.JSONDecodeError:
raise ValueError("Malformed partial JSON encountered.")
@llm_completion_callback()
def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
self.allow_self_signed_https(True)
data = {
"messages": [
{"role": "user", "content": f"Generate output in plain JSON format without any markdown, code block formatting, or additional commentary. Only return the JSON data. {prompt}"}
],
"max_tokens": self.num_output,
"temperature": 0.7,
"top_p": 1
}
headers = {'Content-Type': 'application/json', 'Authorization': ('Bearer ' + self.api_key)}
response_text = self.post_to_azure_api(self.api_url, data, headers)
# Log the raw JSON response
logging.debug(f"Raw JSON response: {response_text}")
# Strip unwanted formatting and extra data
response_text = response_text.replace("```json", "").replace("```", "").strip()
response_text = re.sub(r'[\x00-\x1F\x7F]', '', response_text)
try:
# Parse the response text using the partial JSON parser
response_json = self.parse_partial_json(response_text)
# Extract the content from the response
content = response_json["choices"][0]["message"]["content"]
# Strip unwanted formatting and extra data from the content
content = content.replace("```json", "").replace("```", "").strip()
content = re.sub(r'[\x00-\x1F\x7F]', '', content)
# Parse the content to a list of PageResults objects
pages = self.parse_partial_json(content)
page_objects = []
for page in pages:
# Ensure page is always an integer
page_number = page["page"]
if isinstance(page_number, list):
page_number = page_number[0] if page_number else 0
page_objects.append(PageResults(page=int(page_number), summary=page["summary"]))
# Return the formatted response
return CompletionResponse(text=json.dumps([page.dict() for page in page_objects]))
except json.JSONDecodeError as e:
logging.error(f"JSON decoding error: {e}")
logging.error(f"Response text: {response_text}")
raise eBy using the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu in the How can I modify the phrase_partial_json to avoid errors like the above. |
Beta Was this translation helpful? Give feedback.
-
|
To modify the Here is the modified version of the import json
def parse_partial_json(s):
"""Parse an incomplete JSON string into a valid python dictionary.
NOTE: This is adapted from
https://github.com/OpenInterpreter/open-interpreter/blob/5b6080fae1f8c68938a1e4fa8667e3744084ee21/interpreter/utils/parse_partial_json.py
"""
# Attempt to parse the string as-is.
try:
return json.loads(s)
except json.JSONDecodeError:
pass
# Initialize variables.
new_s = ""
stack = []
is_inside_string = False
escaped = False
# Process each character in the string one at a time.
for char in s:
if is_inside_string:
if char == '"' and not escaped:
is_inside_string = False
elif char == "\n" and not escaped:
char = "\\n" # Replace the newline character with the escape sequence.
elif char == "\\" and not escaped:
escaped = True
else:
escaped = False
else:
if char == '"':
is_inside_string = True
escaped = False
elif char == "{":
stack.append("}")
elif char == "[":
stack.append("]")
elif char == "}" or char == "]":
if stack and stack[-1] == char:
stack.pop()
else:
# Mismatched closing character; the input is malformed.
raise ValueError("Malformed partial JSON encountered.")
# Append the processed character to the new string.
new_s += char
# If we're still inside a string at the end of processing, we need to close the string.
if is_inside_string:
new_s += '"'
# Close any remaining open structures in the reverse order that they were opened.
for closing_char in reversed(stack):
new_s += closing_char
# Attempt to parse the modified string as JSON.
try:
return json.loads(new_s)
except json.JSONDecodeError:
# If we still can't parse the string as JSON, raise error to indicate failure.
raise ValueError("Malformed partial JSON encountered.")This modification ensures that escape sequences, including those for apostrophes, are correctly handled within the string. Additionally, you can refer to the test cases provided in the import pytest
from llama_index.program.openai.utils import parse_partial_json
def test_valid_partial_json():
assert parse_partial_json("{") == {}
with pytest.raises(ValueError):
parse_partial_json('{"foo":')
assert parse_partial_json('{"foo": "bar') == {"foo": "bar"}
def test_invalid_partial_json():
with pytest.raises(ValueError):
parse_partial_json("foo")
def test_partial_json_with_string():
assert parse_partial_json('{"foo": "') == {"foo": ""}
def test_nested_brackets():
assert parse_partial_json('{"foo": [1, 2, 3') == {"foo": [1, 2, 3]}
with pytest.raises(ValueError):
parse_partial_json('{"foo": [1, 2, 3}')
def test_escape_sequences():
assert parse_partial_json('{"foo": "new\\nline"}') == {"foo": "new\nline"}
def test_nested_objects():
assert parse_partial_json('{"foo": {"bar":{') == {"foo": {"bar": {}}}
with pytest.raises(ValueError):
parse_partial_json('{"foo": {"bar":')
def test_mismatched_closing_brackets():
with pytest.raises(ValueError):
parse_partial_json('{"foo": []}}')
def test_complex_partial_json():
assert parse_partial_json(
'{"foo": [1, {"bar": "baz"}, {"qux": ["hello", "world"]}]}'
) == {"foo": [1, {"bar": "baz"}, {"qux": ["hello", "world"]}]}These tests will help ensure that your |
Beta Was this translation helpful? Give feedback.
-
|
@dosu In my retriever output, I have metadata which is |
Beta Was this translation helpful? Give feedback.
-
|
To include the metadata value Here is an example of how you can achieve this:
Below is a code snippet demonstrating these changes: def retrieve_with_metadata(self, str_or_query_bundle: QueryType) -> Tuple[List[Dict[str, Any]], Dict]:
"""Retrieve with metadata."""
if isinstance(str_or_query_bundle, str):
query_bundle = QueryBundle(str_or_query_bundle)
else:
query_bundle = str_or_query_bundle
raw_response_str, metadata = self._sql_database.run_sql(query_bundle.query_str)
if self._return_raw:
nodes_with_metadata = [
{
"page": node.metadata.get('page', 'unknown'),
"answer": node.text
}
for node in [
NodeWithScore(
node=TextNode(
text=raw_response_str,
metadata={
"sql_query": query_bundle.query_str,
"result": metadata["result"],
"col_keys": metadata["col_keys"],
},
excluded_embed_metadata_keys=[
"sql_query",
"result",
"col_keys",
],
excluded_llm_metadata_keys=["sql_query", "result", "col_keys"],
)
).node
]
]
return nodes_with_metadata, metadata
else:
# return formatted
results = metadata["result"]
col_keys = metadata["col_keys"]
formatted_nodes = self._format_node_results(results, col_keys)
nodes_with_metadata = [
{
"page": node.node.metadata.get('page', 'unknown'),
"answer": node.node.text
}
for node in formatted_nodes
]
return nodes_with_metadata, metadataIn this code:
Additionally, you can use from llama_index.core.schema import MetadataMode
def retrieve_with_metadata(self, str_or_query_bundle: QueryType) -> Tuple[List[Dict[str, Any]], Dict]:
"""Retrieve with metadata."""
if isinstance(str_or_query_bundle, str):
query_bundle = QueryBundle(str_or_query_bundle)
else:
query_bundle = str_or_query_bundle
raw_response_str, metadata = self._sql_database.run_sql(query_bundle.query_str)
if self._return_raw:
nodes_with_metadata = [
{
"page": node.metadata.get('page', 'unknown'),
"answer": node.text
}
for node in [
NodeWithScore(
node=TextNode(
text=raw_response_str,
metadata={
"sql_query": query_bundle.query_str,
"result": metadata["result"],
"col_keys": metadata["col_keys"],
},
excluded_embed_metadata_keys=[
"sql_query",
"result",
"col_keys",
],
excluded_llm_metadata_keys=["sql_query", "result", "col_keys"],
)
).node
]
]
return nodes_with_metadata, metadata
else:
# return formatted
results = metadata["result"]
col_keys = metadata["col_keys"]
formatted_nodes = self._format_node_results(results, col_keys)
nodes_with_metadata = [
{
"page": node.node.metadata.get('page', 'unknown'),
"answer": node.node.get_content(MetadataMode.LLM)
}
for node in formatted_nodes
]
return nodes_with_metadata, metadataThis approach ensures that the metadata value |
Beta Was this translation helpful? Give feedback.
-
|
@dosu So how can I include the above in the RetrievealQueryEngine? |
Beta Was this translation helpful? Give feedback.
-
|
To include the
Here is the updated code: from typing import List, Dict, Any
class CustomRetrieverQueryEngine(RetrieverQueryEngine):
def _retrieve(self, query_bundle: QueryBundle) -> List[Dict[str, Any]]:
nodes = self._retriever.retrieve(query_bundle)
nodes = self._apply_node_postprocessors(nodes, query_bundle=query_bundle)
formatted_response = [
{
"page": node.node.metadata.get('page', 'unknown'),
"answer": node.node.text
}
for node in nodes
]
return formatted_response
async def _aretrieve(self, query_bundle: QueryBundle) -> List[Dict[str, Any]]:
nodes = await self._retriever.aretrieve(query_bundle)
nodes = self._apply_node_postprocessors(nodes, query_bundle=query_bundle)
formatted_response = [
{
"page": node.node.metadata.get('page', 'unknown'),
"answer": node.node.text
}
for node in nodes
]
return formatted_responseExample Usage
from pydantic import BaseModel
class Page(BaseModel):
"""Data model for a page."""
page: int
summary: str
from llama_index.legacy.core.base_retriever import BaseRetriever
from llama_index.legacy.query_engine.retriever_query_engine import RetrieverQueryEngine
from llama_index.legacy.callbacks.base import CallbackManager
from llama_index.legacy.response_synthesizers import get_response_synthesizer
# Assuming you have an instance of BaseRetriever
retriever = BaseRetriever()
# Create an instance of CustomRetrieverQueryEngine with the Page class as output_cls
query_engine = CustomRetrieverQueryEngine.from_args(
retriever=retriever,
output_cls=Page,
callback_manager=CallbackManager([]),
response_synthesizer=get_response_synthesizer(service_context=None)
)
query = f"""As an expert in evaluating company annual reports, use only the {company} Annual Report to identify the top 5 pages that best address the following check criteria.
Provide a brief summary (not more than 40 words.) for each page. Focus on the main body of the PDF, prioritizing the Sections listed below.
Exclude resulting any annexures, appendices, glossary, or index sections typically found at the end of the report.
Check criteria details: {criterion}
Answer in JSON format:
[{{
"page": [page_number],
"summary": "[brief_summary]"
}}]
"""
response = query_engine.query(query)
# Assuming response is a list of Page objects
for page in response:
print(page.page)
print(page.summary)By ensuring that the response from your custom LLM is correctly formatted as a JSON string that matches the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu Where is the class |
Beta Was this translation helpful? Give feedback.
-
|
To rerank the combined results from the BM25 retriever and ChromaDB auto retriever to ensure a consistent ranking despite their different scales of scores, you can use the Here is an example of how to set up and use the from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# Initialize document store and add documents
docstore = SimpleDocumentStore()
docstore.add_documents(nodes)
# Initialize ChromaDB and vector store
db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("dense_vectors")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
# Create storage context
storage_context = StorageContext.from_defaults(
docstore=docstore, vector_store=vector_store
)
# Create vector store index
index = VectorStoreIndex(nodes=nodes, storage_context=storage_context)
# Apply nest_asyncio to handle async operations
import nest_asyncio
nest_asyncio.apply()
# Import and set up QueryFusionRetriever
from llama_index.core.retrievers import QueryFusionRetriever, FUSION_MODES
retriever = QueryFusionRetriever(
[
index.as_retriever(similarity_top_k=2),
BM25Retriever.from_defaults(
docstore=index.docstore, similarity_top_k=2
),
],
num_queries=1,
use_async=True,
mode=FUSION_MODES.RECIPROCAL_RANK # Choose the appropriate fusion mode
)
# Retrieve and display nodes
nodes = retriever.retrieve("What happened at Viaweb and Interleaf?")
for node in nodes:
display_source_node(node, source_length=5000)The For more advanced retrieval techniques and comprehensive guides, you can refer to the LlamaIndex documentation, which includes examples on hybrid search, reranking, auto-merging retrieval, and more. Some relevant guides include:
These resources can provide additional insights and examples to help you implement and customize your retrieval and reranking strategies [1][2][3][4]. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu But I'm passing two different query strings to the 2 retrievers. A keyword based query to BM25 and more descriptive query to auto retrieval. In that case if I want to rerank both the outputs, what are my options? |
Beta Was this translation helpful? Give feedback.
-
|
To rerank the outputs from two different retrievers when you are passing different query strings to each (a keyword-based query to BM25 and a more descriptive query to the auto retriever), you can still use the Here’s how you can achieve this:
Step-by-Step Implementation
keyword_query = "Your keyword-based query here"
descriptive_query = "Your more descriptive query here"
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# Initialize document store and add documents
docstore = SimpleDocumentStore()
docstore.add_documents(nodes)
# Initialize ChromaDB and vector store
db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("dense_vectors")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
# Create storage context
storage_context = StorageContext.from_defaults(
docstore=docstore, vector_store=vector_store
)
# Create vector store index
index = VectorStoreIndex(nodes=nodes, storage_context=storage_context)
# Initialize BM25 retriever
from llama_index.retrievers.bm25 import BM25Retriever
bm25_retriever = BM25Retriever.from_defaults(
docstore=index.docstore, similarity_top_k=2
)
# Initialize ChromaDB auto retriever
vector_retriever = index.as_retriever(similarity_top_k=2)
from llama_index.core.retrievers import QueryFusionRetriever, FUSION_MODES
# Apply nest_asyncio to handle async operations
import nest_asyncio
nest_asyncio.apply()
# Define a custom retriever to handle different queries
class CustomQueryFusionRetriever(QueryFusionRetriever):
def retrieve(self, keyword_query: str, descriptive_query: str):
keyword_nodes = self.retrievers[0].retrieve(keyword_query)
descriptive_nodes = self.retrievers[1].retrieve(descriptive_query)
combined_nodes = keyword_nodes + descriptive_nodes
return self._rerank(combined_nodes)
# Initialize the custom query fusion retriever
retriever = CustomQueryFusionRetriever(
[bm25_retriever, vector_retriever],
num_queries=1,
use_async=True,
mode=FUSION_MODES.RECIPROCAL_RANK # Choose the appropriate fusion mode
)
# Retrieve and display nodes
nodes_with_scores = retriever.retrieve(keyword_query, descriptive_query)
for node in nodes_with_scores:
print(f"Page: {node.node.metadata.get('page')}, Score: {node.score:.2f}, Text: {node.node.text[:100]}...")Explanation
This approach allows you to handle different query strings for each retriever and combine their results effectively, leveraging the existing infrastructure of LlamaIndex without extensive customization [1][2][3][4]. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I got the below error trying to use CustomQueryFusionRetriever
|
Beta Was this translation helpful? Give feedback.
-
|
The error you're encountering indicates that the Here’s the corrected implementation of the from llama_index.core.retrievers import QueryFusionRetriever, FUSION_MODES
from llama_index.core.schema import QueryBundle, NodeWithScore
from typing import List
class CustomQueryFusionRetriever(QueryFusionRetriever):
def retrieve(self, keyword_query: str, descriptive_query: str) -> List[NodeWithScore]:
keyword_query_bundle = QueryBundle(keyword_query)
descriptive_query_bundle = QueryBundle(descriptive_query)
keyword_nodes = self._retrievers[0].retrieve(keyword_query_bundle)
descriptive_nodes = self._retrievers[1].retrieve(descriptive_query_bundle)
combined_nodes = keyword_nodes + descriptive_nodes
return self._rerank(combined_nodes)
async def aretrieve(self, keyword_query: str, descriptive_query: str) -> List[NodeWithScore]:
keyword_query_bundle = QueryBundle(keyword_query)
descriptive_query_bundle = QueryBundle(descriptive_query)
keyword_nodes = await self._retrievers[0].aretrieve(keyword_query_bundle)
descriptive_nodes = await self._retrievers[1].aretrieve(descriptive_query_bundle)
combined_nodes = keyword_nodes + descriptive_nodes
return self._rerank(combined_nodes)Example Usage
keyword_query = "Your keyword-based query here"
descriptive_query = "Your more descriptive query here"
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# Initialize document store and add documents
docstore = SimpleDocumentStore()
docstore.add_documents(nodes)
# Initialize ChromaDB and vector store
db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("dense_vectors")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
# Create storage context
storage_context = StorageContext.from_defaults(
docstore=docstore, vector_store=vector_store
)
# Create vector store index
index = VectorStoreIndex(nodes=nodes, storage_context=storage_context)
# Initialize BM25 retriever
from llama_index.retrievers.bm25 import BM25Retriever
bm25_retriever = BM25Retriever.from_defaults(
docstore=index.docstore, similarity_top_k=2
)
# Initialize ChromaDB auto retriever
vector_retriever = index.as_retriever(similarity_top_k=2)
# Apply nest_asyncio to handle async operations
import nest_asyncio
nest_asyncio.apply()
# Initialize the custom query fusion retriever
retriever = CustomQueryFusionRetriever(
[bm25_retriever, vector_retriever],
num_queries=1,
use_async=True,
mode=FUSION_MODES.RECIPROCAL_RANK # Choose the appropriate fusion mode
)
# Retrieve and display nodes
nodes_with_scores = retriever.retrieve(keyword_query, descriptive_query)
for node in nodes_with_scores:
print(f"Page: {node.node.metadata.get('page')}, Score: {node.score:.2f}, Text: {node.node.text[:100]}...")This setup ensures that the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I get the below error.
In the Llamaindex documentation, below is the QueryFusionRetreiver class which we need to customise. `class QueryFusionRetriever(BaseRetriever): |
Beta Was this translation helpful? Give feedback.
-
|
To customize the Here's how you can achieve this:
By following these steps, you can customize the For a more detailed example, you can refer to the LlamaIndex documentation which provides a comprehensive guide on building an advanced fusion retriever from scratch. This includes steps for query generation, performing vector search for each query, and performing fusion to combine and rerank results [1][2][3]. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu can you give me the complete class definition for CustomQueryFusionRetreiver class? |
Beta Was this translation helpful? Give feedback.
-
|
Here is the complete class definition for the from typing import List, Dict, Tuple, Optional
from llama_index.core.retrievers import QueryFusionRetriever, FUSION_MODES
from llama_index.core.schema import QueryBundle, NodeWithScore
class CustomQueryFusionRetriever(QueryFusionRetriever):
def _retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
# Define different queries for each retriever
queries = [
query_bundle, # Original query for BM25
QueryBundle(f"Descriptive: {query_bundle.query_str}") # More descriptive query for ChromaDB
]
if self.use_async:
results = self._run_nested_async_queries(queries)
else:
results = self._run_sync_queries(queries)
if self.mode == FUSION_MODES.RECIPROCAL_RANK:
return self._reciprocal_rerank_fusion(results)[: self.similarity_top_k]
elif self.mode == FUSION_MODES.RELATIVE_SCORE:
return self._relative_score_fusion(results)[: self.similarity_top_k]
elif self.mode == FUSION_MODES.DIST_BASED_SCORE:
return self._relative_score_fusion(results, dist_based=True)[: self.similarity_top_k]
elif self.mode == FUSION_MODES.SIMPLE:
return self._simple_fusion(results)[: self.similarity_top_k]
else:
raise ValueError(f"Invalid fusion mode: {self.mode}")
async def _aretrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
# Define different queries for each retriever
queries = [
query_bundle, # Original query for BM25
QueryBundle(f"Descriptive: {query_bundle.query_str}") # More descriptive query for ChromaDB
]
results = await self._run_async_queries(queries)
if self.mode == FUSION_MODES.RECIPROCAL_RANK:
return self._reciprocal_rerank_fusion(results)[: self.similarity_top_k]
elif self.mode == FUSION_MODES.RELATIVE_SCORE:
return self._relative_score_fusion(results)[: self.similarity_top_k]
elif self.mode == FUSION_MODES.DIST_BASED_SCORE:
return self._relative_score_fusion(results, dist_based=True)[: self.similarity_top_k]
elif self.mode == FUSION_MODES.SIMPLE:
return self._simple_fusion(results)[: self.similarity_top_k]
else:
raise ValueError(f"Invalid fusion mode: {self.mode}")This class extends the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I get the below error.
|
Beta Was this translation helpful? Give feedback.
-
|
The Here’s the corrected implementation of the
Ensure you import the necessary components correctly. from typing import List, Dict, Tuple, Optional
from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.core.schema import QueryBundle, NodeWithScore
from llama_index.core.prompts import PromptTemplate
from llama_index.core.llms import LLMType
from llama_index.core.callbacks import CallbackManager
from llama_index.core.settings import Settings
from llama_index.core.utils import run_async_tasks
import asyncio
Here is the complete class definition: class CustomQueryFusionRetriever(QueryFusionRetriever):
def __init__(
self,
retrievers: List[BaseRetriever],
llm: Optional[LLMType] = None,
query_gen_prompt: Optional[str] = None,
mode: str = "simple",
similarity_top_k: int = 10,
num_queries: int = 4,
use_async: bool = True,
verbose: bool = False,
callback_manager: Optional[CallbackManager] = None,
objects: Optional[List[IndexNode]] = None,
object_map: Optional[dict] = None,
retriever_weights: Optional[List[float]] = None,
) -> None:
self.num_queries = num_queries
self.query_gen_prompt = query_gen_prompt or "Generate {num_queries} queries based on: {query}"
self.similarity_top_k = similarity_top_k
self.mode = mode
self.use_async = use_async
self._retrievers = retrievers
if retriever_weights is None:
self._retriever_weights = [1.0 / len(retrievers)] * len(retrievers)
else:
total_weight = sum(retriever_weights)
self._retriever_weights = [w / total_weight for w in retriever_weights]
self._llm = llm or Settings.llm
super().__init__(
callback_manager=callback_manager,
object_map=object_map,
objects=objects,
verbose=verbose,
)
def _get_queries(self, original_query: str) -> List[str]:
prompt_str = self.query_gen_prompt.format(
num_queries=self.num_queries - 1,
query=original_query,
)
response = self._llm.complete(prompt_str)
queries = response.text.split("\n")
queries = [q.strip() for q in queries if q.strip()]
if self._verbose:
queries_str = "\n".join(queries)
print(f"Generated queries:\n{queries_str}")
return queries[: self.num_queries - 1]
def _retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
queries = [
query_bundle, # Original query for BM25
QueryBundle(f"Descriptive: {query_bundle.query_str}") # More descriptive query for ChromaDB
]
if self.use_async:
results = self._run_nested_async_queries(queries)
else:
results = self._run_sync_queries(queries)
if self.mode == "reciprocal_rank":
return self._reciprocal_rerank_fusion(results)[: self.similarity_top_k]
elif self.mode == "relative_score":
return self._relative_score_fusion(results)[: self.similarity_top_k]
elif self.mode == "dist_based_score":
return self._relative_score_fusion(results, dist_based=True)[: self.similarity_top_k]
elif self.mode == "simple":
return self._simple_fusion(results)[: self.similarity_top_k]
else:
raise ValueError(f"Invalid fusion mode: {self.mode}")
async def _aretrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
queries = [
query_bundle, # Original query for BM25
QueryBundle(f"Descriptive: {query_bundle.query_str}") # More descriptive query for ChromaDB
]
results = await self._run_async_queries(queries)
if self.mode == "reciprocal_rank":
return self._reciprocal_rerank_fusion(results)[: self.similarity_top_k]
elif self.mode == "relative_score":
return self._relative_score_fusion(results)[: self.similarity_top_k]
elif self.mode == "dist_based_score":
return self._relative_score_fusion(results, dist_based=True)[: self.similarity_top_k]
elif self.mode == "simple":
return self._simple_fusion(results)[: self.similarity_top_k]
else:
raise ValueError(f"Invalid fusion mode: {self.mode}")Example Usage
keyword_query = "Your keyword-based query here"
descriptive_query = "Your more descriptive query here"
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# Initialize document store and add documents
docstore = SimpleDocumentStore()
docstore.add_documents(nodes)
# Initialize ChromaDB and vector store
db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("dense_vectors")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
# Create storage context
storage_context = StorageContext.from_defaults(
docstore=docstore, vector_store=vector_store
)
# Create vector store index
index = VectorStoreIndex(nodes=nodes, storage_context=storage_context)
# Initialize BM25 retriever
from llama_index.retrievers.bm25 import BM25Retriever
bm25_retriever = BM25Retriever.from_defaults(
docstore=index.docstore, similarity_top_k=2
)
# Initialize ChromaDB auto retriever
vector_retriever = index.as_retriever(similarity_top_k=2)
# Apply nest_asyncio to handle async operations
import nest_asyncio
nest_asyncio.apply()
# Initialize the custom query fusion retriever
retriever = CustomQueryFusionRetriever(
[bm25_retriever, vector_retriever],
num_queries=1,
use_async=True,
mode="reciprocal_rank" # Choose the appropriate fusion mode
)
# Retrieve and display nodes
nodes_with_scores = retriever.retrieve(keyword_query, descriptive_query)
for node in nodes_with_scores:
print(f"Page: {node.node.metadata.get('page')}, Score: {node.score:.2f}, Text: {node.node.text[:100]}...")This setup ensures that the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu The below is a reranking function. Can we use that in the above case rather customising the retriever? `def reciprocal_rank_fusion(search_results_dict, k=60): |
Beta Was this translation helpful? Give feedback.
-
|
Yes, you can use the provided
Here’s a step-by-step implementation: Step-by-Step Implementation
keyword_query = "Your keyword-based query here"
descriptive_query = "Your more descriptive query here"
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# Initialize document store and add documents
docstore = SimpleDocumentStore()
docstore.add_documents(nodes)
# Initialize ChromaDB and vector store
db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("dense_vectors")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
# Create storage context
storage_context = StorageContext.from_defaults(
docstore=docstore, vector_store=vector_store
)
# Create vector store index
index = VectorStoreIndex(nodes=nodes, storage_context=storage_context)
# Initialize BM25 retriever
from llama_index.retrievers.bm25 import BM25Retriever
bm25_retriever = BM25Retriever.from_defaults(
docstore=index.docstore, similarity_top_k=2
)
# Initialize ChromaDB auto retriever
vector_retriever = index.as_retriever(similarity_top_k=2)
# Apply nest_asyncio to handle async operations
import nest_asyncio
nest_asyncio.apply()
# Retrieve nodes
keyword_nodes = bm25_retriever.retrieve(keyword_query)
descriptive_nodes = vector_retriever.retrieve(descriptive_query)
def reciprocal_rank_fusion(search_results_dict, k=60):
fused_scores = {}
for query, doc_scores in search_results_dict.items():
for rank, (doc, score) in enumerate(sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)):
if doc not in fused_scores:
fused_scores[doc] = 0
fused_scores[doc] += 1 / (rank + k)
reranked_results = {doc: score for doc, score in sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)}
return reranked_results
# Prepare search results dictionary
search_results_dict = {
"keyword_query": {node.node.node_id: node.score for node in keyword_nodes},
"descriptive_query": {node.node.node_id: node.score for node in descriptive_nodes}
}
# Rerank the nodes
reranked_results = reciprocal_rank_fusion(search_results_dict)
# Get the top-ranked nodes
top_ranked_node_ids = list(reranked_results.keys())[:top_k]
top_ranked_nodes = [node for node in keyword_nodes + descriptive_nodes if node.node.node_id in top_ranked_node_ids]
from llama_index.core.llms.openai import OpenAI
# Initialize the LLM
llm = OpenAI(api_key="your_openai_api_key")
# Generate summaries for the top-ranked nodes
for node_with_score in top_ranked_nodes:
node = node_with_score.node
summary = llm.complete(f"Summarize the following content briefly: {node.get_content()}")
print(f"Page: {node.metadata.get('page')}, Summary: {summary.text}")Explanation
This approach leverages the provided |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I have a Chroma DB collection with the metadata field title. How can I add a metadata Filter in Llamaindex in order to filter a list of similar titles matched using regex patterns? An example of what I'm trying is as follows. `unwanted_patterns = [ |
Beta Was this translation helpful? Give feedback.
-
|
The error The relevant code snippet is: response_data = utils.post(self.base_url, headers=self.headers, params=params)
nodes = []
for ele in response_data["nodes"]:
text_node = TextNode.parse_obj(ele["node"])
nodes.append(NodeWithScore(node=text_node, score=ele["score"]))The error occurs because the code attempts to access To resolve this issue, you should ensure that all expected keys are present in the For more details, you can refer to the source file: retriever.py [1]. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I have the following function to create a Chromadb in Llamaindex. However, in What could be the reason and how to correct it? |
Beta Was this translation helpful? Give feedback.
-
|
The error Here is the corrected function with the necessary changes to set the appropriate permissions using import os
import re
import chromadb
from llama_index.core import StorageContext, VectorStoreIndex
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core.readers import Document
def get_chromadbindex(company):
persist_dir = os.path.expanduser(f'chromadb/{company}')
if not os.path.exists(persist_dir):
print(f"{persist_dir} Does not exist. Creating ...")
os.makedirs(persist_dir, exist_ok=True)
os.chmod(persist_dir, 0o755) # Ensure the directory has write permissions
result = parse_and_load_report(company)
documents = [Document(text=result["content"], metadata={"page": result["page"], "titles": result["titles"]}) for result in result]
collection = re.sub(r"\s+", "_", company)
db = chromadb.PersistentClient(path=persist_dir)
chroma_collection = db.get_or_create_collection(f"{collection}")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection, collection_name=collection, persist_dir=persist_dir)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
chroma_index = VectorStoreIndex.from_documents(documents, storage_context=storage_context, embed_model=embed_model, show_progress=True)
collection = re.sub(r"\s+", "_", company)
db = chromadb.PersistentClient(path=persist_dir)
chroma_collection = db.get_or_create_collection(f"{collection}")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
chroma_index = VectorStoreIndex(vector_store, embed_model=embed_model, show_progress=True)
return chroma_indexExplanation
By ensuring that the directory has the correct write permissions, you should be able to avoid the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I am trying to use Llamaindex with ChromaDB vectorstore, where I want to apply a metadata filter based on the page numbers which is an integer in my vectorstore. I have a list of page numbers called pages where I want to filter only those pages in the list to be retrieved. However, when use it as a list of integers, I got the below error. Which resulted the below error. |
Beta Was this translation helpful? Give feedback.
-
|
To apply a metadata filter based on a list of page numbers (integers) in LlamaIndex with ChromaDB vector store and avoid the error indicating that the value is not a valid integer or float, you need to ensure that the filter values are correctly formatted as integers. Here is how you can do it:
Here is an example of how you can apply a metadata filter based on a list of page numbers: from llama_index.vector_stores.chroma.base import ChromaVectorStore, MetadataFilters, MetadataFilter
from llama_index.vector_stores.vertexaivectorsearch.utils import to_vectorsearch_filter
# Define your metadata filters
page_numbers = [258, 39, 43, 77, 14, 113, 25, 277, 23, 89, 94, 95] # List of page numbers (integers)
filters = MetadataFilters(
filters=[
MetadataFilter(key="page", operator="in", value=page_numbers)
]
)
# Convert to Chroma-specific filters
txt_filters, num_filters = to_vectorsearch_filter(filters)
# Initialize your ChromaVectorStore
chroma_vector_store = ChromaVectorStore.from_params(
collection_name="your_collection_name",
persist_dir="your_persist_dir"
)
# Apply the filters to get nodes
nodes = chroma_vector_store.get_nodes(node_ids=None, filters=filters)In this example, the This approach ensures that the filter values are valid integers, avoiding the error indicating that the value is not a valid integer or float. Additionally, the use of |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I have a set of nodes retrieved from a vectorstore where I want to answer a question from each node iteratively. How can I improve the efficiency of this using Llamaindex? Is there any method to generate answers without sending it to an LLM API? |
Beta Was this translation helpful? Give feedback.
-
|
To truncate a node to a fixed number of tokens before inputting it to a query engine in LlamaIndex, you can use the from llama_index.core.indices.utils import truncate_text
from llama_index.legacy.node_parser.text import TokenTextSplitter
from llama_index.core.schema import BaseNode
# Assuming you have a node and a TokenTextSplitter instance
node = BaseNode() # Replace with your actual node
text_splitter = TokenTextSplitter(max_tokens=100) # Set the max tokens you want
# Get the content of the node
node_text = " ".join(node.get_content().splitlines())
# Truncate the text to the specified number of tokens
truncated_text = truncate_text(node_text, text_splitter)
# Now you can input the truncated text to your query engine
print(truncated_text)In this example, |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I'm trying to setup my Embedding Model in Llamaindex as below: However, I got the below error. |
Beta Was this translation helpful? Give feedback.
-
|
The error you're encountering, To resolve this issue, you need to ensure that the Correct Setup Examplefrom llama_index.embeddings.langchain import LangchainEmbedding
from langchain.embeddings import HuggingFaceBgeEmbeddings
# Initialize the HuggingFaceBgeEmbeddings
hf_embeddings = HuggingFaceBgeEmbeddings(model_name="BAAI/bge-base-en")
# Initialize the LangchainEmbedding with the HuggingFaceBgeEmbeddings
embedding_model = LangchainEmbedding(langchain_embeddings=hf_embeddings)
# Set the embedding model in the Settings
Settings.embed_model = embedding_model
# Now you can use embedding_model for your tasks
query_embedding = embedding_model._get_query_embedding("your query")
text_embedding = embedding_model._get_text_embedding("your text")Explanation
Additional Example Using
|
Beta Was this translation helpful? Give feedback.
-
|
@dosu Hi, I have the below function which runs parallel.
I can directly call the process_company and run it without any error. However, when I try to run in parallel, I got the following error. Can you suggest a way to set my LLM when running in parallel? |
Beta Was this translation helpful? Give feedback.
-
|
To set your LLM when running the
Here’s an example of how you can achieve this: Step-by-Step Implementation
Ensure that the def process_company(row, llm):
# Set the LLM for the current process
Settings.llm = llm
# Your processing logic here
# Example: print(f"Processing company: {row['Company']} with LLM: {Settings.llm}")
# Replace with your actual processing logic
pass
Use import pandas as pd
from tqdm import tqdm
from datetime import datetime
from joblib import Parallel, delayed
from llama_index.core import Settings
# Initialize your custom LLM
llm = AzureHostedLLM()
Settings.llm = llm
def run_all_companies():
# Example data
data = {'ID': [159, 160], 'Company': ['National Development Bank PLC', 'Softlogic Life Insurance PLC']}
new_companies = pd.DataFrame(data)
if new_companies.empty:
return f"No new companies found."
# Use joblib to parallelize the process_company function
Parallel(n_jobs=4)(delayed(process_company)(row, llm) for _, row in tqdm(new_companies.iterrows(), total=new_companies.shape[0]))
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
return f"Execution Cycle Completed at {timestamp} with {len(new_companies)} companies"
# Example usage
if __name__ == "__main__":
result = run_all_companies()
print(result)Explanation
By following these steps, you ensure that the LLM is correctly initialized in each parallel process, avoiding the error related to the missing OpenAI API key. This approach leverages |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I am trying to create a QASummaryQueryEngineBuilder using a Chromadb vectorstore. My Code is as below. However I got an error in running a query. |
Beta Was this translation helpful? Give feedback.
-
|
To create a Step-by-Step Implementation
Make sure the def get_chromadbindex(company):
persist_dir = os.path.expanduser(f'chromadb/{company}')
if not os.path.exists(persist_dir):
print(f"{persist_dir} Does not exist. Creating ...")
os.makedirs(persist_dir, exist_ok=True)
os.chmod(persist_dir, 0o755) # Ensure the directory has write permissions
result = parse_and_load_report(company)
documents = [Document(text=result["content"], metadata={"page": result["page"], "titles": result["titles"]}) for result in result]
collection = re.sub(r"\s+", "_", company)
db = chromadb.PersistentClient(path=persist_dir)
chroma_collection = db.get_or_create_collection(f"{collection}")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection, collection_name=collection, persist_dir=persist_dir)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
chroma_index = VectorStoreIndex.from_documents(documents, storage_context=storage_context, embed_model=embed_model, show_progress=True)
collection = re.sub(r"\s+", "_", company)
db = chromadb.PersistentClient(path=persist_dir)
chroma_collection = db.get_or_create_collection(f"{collection}")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
chroma_index = VectorStoreIndex(vector_store, embed_model=embed_model, show_progress=True)
return chroma_index
Ensure you are correctly creating the from llama_index.core.composability import QASummaryQueryEngineBuilder
# Get the vector store index
index = get_chromadbindex('National Development Bank PLC')
# Ensure the index is not None
if index is None:
raise ValueError("Failed to get the vector store index")
# Create the query engine builder
query_engine_builder = QASummaryQueryEngineBuilder(
llm=llm,
)
# Build the query engine from documents
query_engine = query_engine_builder.build_from_documents(index.docstore.docs.values())Troubleshooting Common ErrorsIf you encounter an error when running a query, it might be due to the following reasons:
Example Query# Example query
response = query_engine.query("What did the author do growing up?")
print(response)By following these steps, you should be able to create a |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I'm using a node.text output which is in markdown text format. I want to input that into a query engine as context and then generate a summary of it. Rather I input that markdown text in row format, are there any preprocessing I need to do and what is the simplest and best option to get it done? |
Beta Was this translation helpful? Give feedback.
-
|
To preprocess markdown text from Here is a simple example of how you can achieve this: from llama_index.legacy.schema import TextNode
from llama_index.legacy.extractors.metadata_extractors import SummaryExtractor
from llama_index.legacy.llms.llm import LLM
# Initialize the LLM (replace 'your_llm' with the actual LLM you are using)
llm = LLM(name='your_llm')
# Create a TextNode with your markdown text
node = TextNode(text="Your markdown text here")
# Initialize the SummaryExtractor
summary_extractor = SummaryExtractor(llm=llm, summaries=["self"])
# Extract the summary
summary = await summary_extractor.aextract([node])
# The summary will be in the 'section_summary' field of the result
print(summary[0]["section_summary"])This code initializes a |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I'm using Llamaindex RAG with Python Joblib Parallel. I am using a custom LLM and local embed model. However, when run in parallel the Settings LLM and embed_model does not recognised and give an error to set embed_model and llm. How can I use these Llamaindex settings with parallel with joblib? |
Beta Was this translation helpful? Give feedback.
-
|
To use LlamaIndex settings with a custom LLM and local embed model in parallel using Python Joblib, you need to ensure that the Step-by-Step Implementation
Ensure that your custom LLM and embedding model are correctly initialized. from llama_index.core.llms.utils import resolve_llm
from llama_index.core.embeddings.utils import resolve_embed_model
# Define your custom LLM and embedding model
custom_llm = resolve_llm("path_to_custom_llm")
local_embed_model = resolve_embed_model("path_to_local_embed_model")
Create a from llama_index.core.service_context import ServiceContext
service_context = ServiceContext.from_defaults(
llm=custom_llm,
embed_model=local_embed_model
)
Ensure that the from joblib import Parallel, delayed
from llama_index.core.service_context import ServiceContext
def process_data(data, service_context):
# Ensure the settings are recognized in each parallel process
ServiceContext.set_global_service_context(service_context)
# Your processing logic here
# For example, use service_context.llm and service_context.embed_model
llm_result = service_context.llm.process(data)
embed_result = service_context.embed_model.embed(data)
return llm_result, embed_result
Use Joblib to process the data in parallel, ensuring that each process recognizes the custom settings. # Example data to process
data_list = ["data1", "data2", "data3"]
# Use Joblib to process data in parallel
results = Parallel(n_jobs=3)(delayed(process_data)(data, service_context) for data in data_list)
print(results)Full ExampleHere is the complete example: from joblib import Parallel, delayed
from llama_index.core.service_context import ServiceContext
from llama_index.core.llms.utils import resolve_llm
from llama_index.core.embeddings.utils import resolve_embed_model
# Define your custom LLM and embedding model
custom_llm = resolve_llm("path_to_custom_llm")
local_embed_model = resolve_embed_model("path_to_local_embed_model")
# Create a ServiceContext with the custom LLM and embedding model
service_context = ServiceContext.from_defaults(
llm=custom_llm,
embed_model=local_embed_model
)
def process_data(data, service_context):
# Ensure the settings are recognized in each parallel process
ServiceContext.set_global_service_context(service_context)
# Your processing logic here
# For example, use service_context.llm and service_context.embed_model
llm_result = service_context.llm.process(data)
embed_result = service_context.embed_model.embed(data)
return llm_result, embed_result
# Example data to process
data_list = ["data1", "data2", "data3"]
# Use Joblib to process data in parallel
results = Parallel(n_jobs=3)(delayed(process_data)(data, service_context) for data in data_list)
print(results)This setup ensures that the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu I got the below error when trying to create an index in Llamaindex with the Joblib parallel process. generate_results function is as follows. How can I set the embedding model in this case to work with Joblib parallel? |
Beta Was this translation helpful? Give feedback.
-
|
To set the embedding model to work with Joblib parallel processing in LlamaIndex, you need to ensure that the embedding model and other settings are correctly initialized in each parallel process. This can be achieved by passing the embedding model and other necessary settings to each worker process. Here’s how you can modify your Step-by-Step Implementation
Ensure that the embedding model is correctly initialized in each parallel process. from llama_index.core import Settings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.vector_stores import MetadataFilters, MetadataFilter, FilterOperator
from llama_index.core.postprocessor.optimizer import SentenceEmbeddingOptimizer
from joblib import Parallel, delayed
import pandas as pd
import os
@joblib.wrap_non_picklable_objects
def generate_results(company, index, row):
# Initialize the embedding model
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-base-en")
Settings.embed_model = embed_model
try:
csv_file = os.path.expanduser(f'~/airflow/dags/results/{company}_v4.csv')
number = row['Numbering']
chkl_ct_id = row['chkl_ct_id']
question = row['Question']
description = row['description'] if pd.notna(row['description']) else ''
pages = row.get('pages')
if pages:
filters = MetadataFilters(
filters=[MetadataFilter(key="page", operator=FilterOperator.IN, value=[int(page) for page in pages])]
)
retriever = index.as_retriever(filters=filters, similarity_top_k=5)
else:
retriever = index.as_retriever(similarity_top_k=5)
query = f"{question} {description}"
nodes = retriever.retrieve(query)
postprocessor = SentenceEmbeddingOptimizer(threshold_cutoff=0.7, embed_model=embed_model)
nodes = postprocessor.postprocess_nodes(nodes=nodes, query_str=query)
summary_df = pd.DataFrame(Parallel(n_jobs=4)(delayed(process_node)(node, question, number) for node in nodes))
summary_df['chkl_ct_id'] = chkl_ct_id
summary_df['company_name'] = company
summary_df['question'] = question
summary_df['description'] = description
summary_df = summary_df[['company_name', 'chkl_ct_id', 'question', 'description', 'summary', 'page']]
summary_df.to_csv(csv_file, mode='a', index=False, header=not os.path.exists(csv_file))
except Exception as e:
raise RuntimeError(f"Error in generating results: {str(e)}")
Ensure that the import pandas as pd
from tqdm import tqdm
from datetime import datetime
from joblib import Parallel, delayed
def run_all_companies():
# Example data
data = {'ID': [159, 160], 'Company': ['National Development Bank PLC', 'Softlogic Life Insurance PLC']}
new_companies = pd.DataFrame(data)
if new_companies.empty:
return f"No new companies found."
# Use joblib to parallelize the generate_results function
Parallel(n_jobs=4)(delayed(generate_results)(row['Company'], index, row) for _, row in tqdm(new_companies.iterrows(), total=new_companies.shape[0]))
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
return f"Execution Cycle Completed at {timestamp} with {len(new_companies)} companies"
# Example usage
if __name__ == "__main__":
result = run_all_companies()
print(result)Explanation
By following these steps, you ensure that the embedding model and other settings are correctly initialized and recognized in each parallel process, avoiding the error related to the missing OpenAI API key. This approach leverages |
Beta Was this translation helpful? Give feedback.
-
Hi I'm using a RetrieverQueryEngine with custom LLM in Llamaindex. The custom LLM is based on an Azure Hosted LLM. I want to use Pydantic class called Page as the output_cls inside the query engine or in the response synthesizer.
query_engine = RetrieverQueryEngine.from_args( output_cls=Page, retriever= hybrid_retriever, llm = llm, node_postprocessors = [lim_reorder,reranker], response_synthesizer=response_synthesizer, )`from pydantic import BaseModel
class Page(BaseModel):
"""Data model for page number and summary"""
Beta Was this translation helpful? Give feedback.
All reactions