Skip to content

Custom

CustomQueryEngine #

Bases: BaseModel, BaseQueryEngine

自定义查询引擎。

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

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

Source code in 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
class CustomQueryEngine(BaseModel, BaseQueryEngine):
    """自定义查询引擎。

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

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

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

    def _get_prompt_modules(self) -> PromptMixinType:
        """获取提示子模块。"""
        return {}

    class Config:
        arbitrary_types_allowed = True

    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:
        """运行自定义查询。"""

    async def acustom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
        """以异步方式运行自定义查询。"""
        # 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

运行自定义查询。

Source code in llama_index/core/query_engine/custom.py 
62
63
64
@abstractmethod
def custom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
    """运行自定义查询。"""

acustom_query async #

acustom_query(query_str: str) -> STR_OR_RESPONSE_TYPE

以异步方式运行自定义查询。

Source code in llama_index/core/query_engine/custom.py 
66
67
68
69
async def acustom_query(self, query_str: str) -> STR_OR_RESPONSE_TYPE:
    """以异步方式运行自定义查询。"""
    # by default, just run the synchronous version
    return self.custom_query(query_str)