跳到主内容

自定义

CustomQueryEngine #

基类：BaseModel, BaseQueryEngine

自定义查询引擎。

子类可以将附加属性定义为 Pydantic 字段。子类必须实现 custom_query 方法，该方法接受查询字符串并返回 Response 对象或字符串作为输出。

它们可以选择实现 acustom_query 方法以支持异步。

参数

名称 类型 描述 默认值
callback_manager CallbackManager
<llama_index.core.callbacks.base.CallbackManager object at 0x7e52a7f323f0>
源代码位于 llama-index-core/llama_index/core/query_engine/custom.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class CustomQueryEngine(BaseModel, BaseQueryEngine):
    """
    Custom query engine.

    Subclasses can define additional attributes as Pydantic fields.
    Subclasses must implement the `custom_query` method, which takes a query string
    and returns either a Response object or a string as output.

    They can optionally implement the `acustom_query` method for async support.

    """

    model_config = ConfigDict(arbitrary_types_allowed=True)
    callback_manager: CallbackManager = Field(
        default_factory=lambda: CallbackManager([]), exclude=True
    )

    def _get_prompt_modules(self) -> PromptMixinType:
        """Get prompt sub-modules."""
        return {}

    def query(self, str_or_query_bundle: QueryType) -> RESPONSE_TYPE:
        with self.callback_manager.as_trace("query"):
            # if query bundle, just run the query
            if isinstance(str_or_query_bundle, QueryBundle):
                query_str = str_or_query_bundle.query_str
            else:
                query_str = str_or_query_bundle
            raw_response = self.custom_query(query_str)
            return (
                Response(raw_response)
                if isinstance(raw_response, str)
                else raw_response
            )

    async def aquery(self, str_or_query_bundle: QueryType) -> RESPONSE_TYPE:
        with self.callback_manager.as_trace("query"):
            if isinstance(str_or_query_bundle, QueryBundle):
                query_str = str_or_query_bundle.query_str
            else:
                query_str = str_or_query_bundle
            raw_response = await self.acustom_query(query_str)
            return (
                Response(raw_response)
                if isinstance(raw_response, str)
                else raw_response
            )

    @abstractmethod
    def custom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
        """Run a custom query."""

    async def acustom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
        """Run a custom query asynchronously."""
        # by default, just run the synchronous version
        return self.custom_query(query_str)

    def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        raise NotImplementedError("This query engine does not support _query.")

    async def _aquery(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        raise NotImplementedError("This query engine does not support _aquery.")

custom_query abstractmethod #

custom_query(query_str: str) -> STR_OR_RESPONSE_TYPE

运行自定义查询。

源代码位于 llama-index-core/llama_index/core/query_engine/custom.py 
64
65
66
@abstractmethod
def custom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
    """Run a custom query."""

acustom_query async #

acustom_query(query_str: str) -> STR_OR_RESPONSE_TYPE

异步运行自定义查询。

源代码位于 llama-index-core/llama_index/core/query_engine/custom.py 
68
69
70
71
async def acustom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
    """Run a custom query asynchronously."""
    # by default, just run the synchronous version
    return self.custom_query(query_str)