Files
wehub-resource-sync c56bef871b
CodeQL / Analyze (python) (push) Has been cancelled
Update Platform Components Table / update (push) Has been cancelled
Docker image release / Build base image (push) Has been cancelled
Sync docs with Docusaurus / sync (push) Has been cancelled
Tests / Check if changed (push) Has been cancelled
Tests / format (push) Has been cancelled
Tests / check-imports (push) Has been cancelled
Tests / Unit / macos-latest (push) Has been cancelled
Tests / Unit / ubuntu-latest (push) Has been cancelled
Tests / Unit / windows-latest (push) Has been cancelled
Tests / mypy (push) Has been cancelled
Tests / Integration / ubuntu-latest (push) Has been cancelled
Tests / Integration / macos-latest (push) Has been cancelled
Tests / Integration / windows-latest (push) Has been cancelled
Tests / notify-slack-on-failure (push) Has been cancelled
Tests / Mark tests as completed (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:22:28 +08:00

404 lines
15 KiB
Python

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from typing import Any
from haystack import Document, component, default_from_dict, default_to_dict, logging
from haystack.components.builders import PromptBuilder
from haystack.components.generators.chat.openai import OpenAIChatGenerator
from haystack.components.generators.chat.types import ChatGenerator
from haystack.core.serialization import component_to_dict
from haystack.dataclasses import ChatMessage
from haystack.utils import deserialize_chatgenerator_inplace
from haystack.utils.async_utils import _execute_component_async
from haystack.utils.misc import _deduplicate_documents, _parse_dict_from_json
logger = logging.getLogger(__name__)
def _default_openai_chat_generator() -> ChatGenerator:
return OpenAIChatGenerator(
model="gpt-4.1-mini",
generation_kwargs={
"temperature": 0.0,
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "document_ranking",
"schema": {
"type": "object",
"properties": {
"documents": {
"type": "array",
"items": {
"type": "object",
"properties": {"index": {"type": "integer"}},
"required": ["index"],
"additionalProperties": False,
},
}
},
"required": ["documents"],
"additionalProperties": False,
},
},
},
},
)
DEFAULT_PROMPT_TEMPLATE = """
You are ranking retrieved documents for relevance to a query.
Return valid JSON only, with this structure:
{
"documents": [
{"index": 1}
]
}
Rules:
- Rank documents from most relevant to least relevant for answering the query.
- Only include documents that are relevant to the query.
- Do not return or rank documents that are not relevant.
- If none are relevant, return {"documents": []}.
- Use only document indices from the provided documents.
- Do not repeat document indices.
- Do not include explanations or any text outside the JSON object.
Query:
{{ query }}
Documents:
{% for document in documents %}
Document {{ loop.index }}:
content: {{ document.content or "" }}
{% endfor %}
""".strip()
@component
class LLMRanker:
"""
Ranks documents for a query using a Large Language Model.
The LLM is expected to return a JSON object containing ranked document indices.
Usage example:
```python
from haystack import Document
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.rankers import LLMRanker
chat_generator = OpenAIChatGenerator(
model="gpt-4.1-mini",
generation_kwargs={
"temperature": 0.0,
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "document_ranking",
"schema": {
"type": "object",
"properties": {
"documents": {
"type": "array",
"items": {
"type": "object",
"properties": {"index": {"type": "integer"}},
"required": ["index"],
"additionalProperties": False,
},
}
},
"required": ["documents"],
"additionalProperties": False,
},
},
},
},
)
ranker = LLMRanker(chat_generator=chat_generator)
documents = [
Document(id="paris", content="Paris is the capital of France."),
Document(id="berlin", content="Berlin is the capital of Germany."),
]
result = ranker.run(query="capital of Germany", documents=documents)
print(result["documents"][0].id)
```
"""
def __init__(
self,
*,
chat_generator: ChatGenerator | None = None,
prompt: str = DEFAULT_PROMPT_TEMPLATE,
top_k: int = 10,
raise_on_failure: bool = False,
) -> None:
"""
Initialize the LLMRanker component.
:param chat_generator:
The chat generator to use for reranking. If `None`, a default `OpenAIChatGenerator` configured for JSON
output is used.
:param prompt:
Custom prompt template for reranking. The prompt must include exactly the variables `query` and
`documents` and instruct the LLM to return ranked 1-based document indices as JSON.
:param top_k:
The maximum number of documents to return.
:param raise_on_failure:
If `True`, raise when generation or response parsing fails. If `False`, log the failure and return the
input documents in fallback order.
"""
if top_k <= 0:
raise ValueError(f"top_k must be > 0, but got {top_k}")
self.top_k = top_k
self.raise_on_failure = raise_on_failure
self.prompt = prompt
self._prompt_builder = PromptBuilder(template=self.prompt, required_variables=["documents", "query"])
if set(self._prompt_builder.variables) != {"documents", "query"}:
raise ValueError("prompt must include exactly the variables 'documents' and 'query'.")
if chat_generator is None:
self._chat_generator = _default_openai_chat_generator()
else:
self._chat_generator = chat_generator
def warm_up(self) -> None:
"""Warm up the underlying chat generator."""
if hasattr(self._chat_generator, "warm_up"):
self._chat_generator.warm_up()
async def warm_up_async(self) -> None:
"""Warm up the underlying chat generator on the serving event loop."""
if hasattr(self._chat_generator, "warm_up_async"):
await self._chat_generator.warm_up_async()
elif hasattr(self._chat_generator, "warm_up"):
self._chat_generator.warm_up()
def close(self) -> None:
"""Release the underlying chat generator's resources."""
if hasattr(self._chat_generator, "close"):
self._chat_generator.close()
async def close_async(self) -> None:
"""Release the underlying chat generator's async resources."""
if hasattr(self._chat_generator, "close_async"):
await self._chat_generator.close_async()
elif hasattr(self._chat_generator, "close"):
self._chat_generator.close()
def to_dict(self) -> dict[str, Any]:
"""
Serialize this component to a dictionary.
:returns:
Dictionary with serialized data.
"""
return default_to_dict(
self,
chat_generator=component_to_dict(obj=self._chat_generator, name="chat_generator"),
prompt=self.prompt,
top_k=self.top_k,
raise_on_failure=self.raise_on_failure,
)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "LLMRanker":
"""
Deserialize this component from a dictionary.
:param data:
The dictionary representation of the component.
:returns:
The deserialized component instance.
"""
init_params = data.get("init_parameters", {})
if init_params.get("chat_generator"):
deserialize_chatgenerator_inplace(init_params, key="chat_generator")
return default_from_dict(cls, data)
@component.output_types(documents=list[Document])
def run(self, query: str, documents: list[Document], top_k: int | None = None) -> dict[str, list[Document]]:
"""
Rank documents for a query using an LLM.
Before ranking, duplicate documents are removed.
:param query:
The query used for reranking.
:param documents:
Candidate documents to rerank.
:param top_k:
The maximum number of documents to return. Overrides the instance's `top_k` if provided.
:returns:
A dictionary with the ranked documents under the `documents` key.
"""
if top_k is not None and top_k <= 0:
raise ValueError(f"top_k must be > 0, but got {top_k}")
if not documents:
return {"documents": []}
top_k = self.top_k if top_k is None else top_k
deduplicated_documents = _deduplicate_documents(documents)
fallback_documents = deduplicated_documents
if not query.strip():
logger.warning("Empty query provided to LLMRanker. Returning documents without reranking.")
return {"documents": fallback_documents}
self.warm_up()
prompt = self._prompt_builder.run(query=query.strip(), documents=deduplicated_documents)
try:
result = self._chat_generator.run(messages=[ChatMessage.from_user(prompt["prompt"])])
except Exception as exc:
if self.raise_on_failure:
raise
logger.warning(
"LLMRanker failed during chat generation. Returning fallback order. Error: {error}", error=exc
)
return {"documents": fallback_documents}
try:
reply_text = self._get_reply_text(result)
ranked_documents = self._rank_documents_from_reply(reply_text=reply_text, documents=deduplicated_documents)
except (TypeError, ValueError) as exc:
if self.raise_on_failure:
raise
logger.warning(
"LLMRanker failed while processing the chat response. Returning fallback order. Error: {error}",
error=exc,
)
return {"documents": fallback_documents}
return {"documents": ranked_documents[:top_k]}
@component.output_types(documents=list[Document])
async def run_async(
self, query: str, documents: list[Document], top_k: int | None = None
) -> dict[str, list[Document]]:
"""
Asynchronously rank documents for a query using an LLM.
Before ranking, duplicate documents are removed.
This is the asynchronous version of the `run` method. It has the same parameters and return values
but can be used with `await` in an async code. If the chat generator only implements a synchronous
`run` method, it is executed in a thread to avoid blocking the event loop.
:param query:
The query used for reranking.
:param documents:
Candidate documents to rerank.
:param top_k:
The maximum number of documents to return. Overrides the instance's `top_k` if provided.
:returns:
A dictionary with the ranked documents under the `documents` key.
"""
if top_k is not None and top_k <= 0:
raise ValueError(f"top_k must be > 0, but got {top_k}")
if not documents:
return {"documents": []}
top_k = self.top_k if top_k is None else top_k
deduplicated_documents = _deduplicate_documents(documents)
fallback_documents = deduplicated_documents
if not query.strip():
logger.warning("Empty query provided to LLMRanker. Returning documents without reranking.")
return {"documents": fallback_documents}
await self.warm_up_async()
prompt = self._prompt_builder.run(query=query.strip(), documents=deduplicated_documents)
try:
result = await _execute_component_async(
self._chat_generator, messages=[ChatMessage.from_user(prompt["prompt"])]
)
except Exception as exc:
if self.raise_on_failure:
raise
logger.warning(
"LLMRanker failed during chat generation. Returning fallback order. Error: {error}", error=exc
)
return {"documents": fallback_documents}
try:
reply_text = self._get_reply_text(result)
ranked_documents = self._rank_documents_from_reply(reply_text=reply_text, documents=deduplicated_documents)
except (TypeError, ValueError) as exc:
if self.raise_on_failure:
raise
logger.warning(
"LLMRanker failed while processing the chat response. Returning fallback order. Error: {error}",
error=exc,
)
return {"documents": fallback_documents}
return {"documents": ranked_documents[:top_k]}
@staticmethod
def _get_reply_text(result: dict[str, Any]) -> str:
replies = result.get("replies") or []
if not replies:
raise ValueError("ChatGenerator returned no replies.")
reply_text = replies[0].text
if reply_text is None:
raise ValueError("ChatGenerator returned a reply without text.")
return reply_text
@staticmethod
def _rank_documents_from_reply(reply_text: str, documents: list[Document]) -> list[Document]:
parsed_response = _parse_dict_from_json(reply_text, expected_keys=["documents"], raise_on_failure=True)
ranked_entries = parsed_response["documents"]
if not isinstance(ranked_entries, list):
raise TypeError("Expected 'documents' in ranking response to be a list.")
if not ranked_entries:
return []
ranked_documents: list[Document] = []
for entry in ranked_entries:
if not isinstance(entry, dict):
raise TypeError("Expected each ranked document entry to be a JSON object.")
document_index = entry.get("index")
if document_index is None:
continue
try:
# LLMs can return numeric indices as strings even when asked for integers.
document_index = int(document_index)
except (TypeError, ValueError):
continue
# Jinja's `loop.index` is 1-based:
# https://jinja.palletsprojects.com/en/stable/templates/#for
if document_index < 1 or document_index > len(documents):
continue
document = documents[document_index - 1]
ranked_documents.append(document)
if not ranked_documents:
raise ValueError("Ranking response did not contain any valid document indices.")
return ranked_documents