Bases: BaseToolSpec
Notion工具规范。
目前是数据加载器的简单封装。
TODO: 添加更多的方法到Notion规范中。
Source code in llama_index/tools/notion/base.py
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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98 | class NotionToolSpec(BaseToolSpec):
"""Notion工具规范。
目前是数据加载器的简单封装。
TODO: 添加更多的方法到Notion规范中。"""
spec_functions = ["load_data", "search_data"]
def __init__(self, integration_token: Optional[str] = None) -> None:
"""使用参数进行初始化。"""
self.reader = NotionPageReader(integration_token=integration_token)
def get_fn_schema_from_fn_name(
self, fn_name: str, spec_functions: Optional[List[SPEC_FUNCTION_TYPE]] = None
) -> Optional[Type[BaseModel]]:
"""从函数名称返回映射。"""
if fn_name == "load_data":
return NotionLoadDataSchema
elif fn_name == "search_data":
return NotionSearchDataSchema
else:
raise ValueError(f"Invalid function name: {fn_name}")
def load_data(
self, page_ids: Optional[List[str]] = None, database_id: Optional[str] = None
) -> str:
"""从一组页面ID或数据库ID加载内容。
如果不知道页面ID或数据库ID，请不要使用此端点。
"""
page_ids = page_ids or []
docs = self.reader.load_data(page_ids=page_ids, database_id=database_id)
return "\n".join([doc.get_content() for doc in docs])
def search_data(
self,
query: str,
direction: Optional[str] = None,
timestamp: Optional[str] = None,
value: Optional[str] = None,
property: Optional[str] = None,
page_size: int = 100,
) -> str:
"""搜索相关页面的列表。
包含每个页面的元数据（但不包括页面内容）。
"""
payload: Dict[str, Any] = {
"query": query,
"page_size": page_size,
}
if direction is not None or timestamp is not None:
payload["sort"] = {}
if direction is not None:
payload["sort"]["direction"] = direction
if timestamp is not None:
payload["sort"]["timestamp"] = timestamp
if value is not None or property is not None:
payload["filter"] = {}
if value is not None:
payload["filter"]["value"] = value
if property is not None:
payload["filter"]["property"] = property
response = requests.post(SEARCH_URL, json=payload, headers=self.reader.headers)
response_json = response.json()
return response_json["results"]
|
get_fn_schema_from_fn_name(
fn_name: str,
spec_functions: Optional[
List[SPEC_FUNCTION_TYPE]
] = None,
) -> Optional[Type[BaseModel]]
从函数名称返回映射。
Source code in llama_index/tools/notion/base.py
43
44
45
46
47
48
49
50
51
52 | def get_fn_schema_from_fn_name(
self, fn_name: str, spec_functions: Optional[List[SPEC_FUNCTION_TYPE]] = None
) -> Optional[Type[BaseModel]]:
"""从函数名称返回映射。"""
if fn_name == "load_data":
return NotionLoadDataSchema
elif fn_name == "search_data":
return NotionSearchDataSchema
else:
raise ValueError(f"Invalid function name: {fn_name}")
|
load_data(
page_ids: Optional[List[str]] = None,
database_id: Optional[str] = None,
) -> str
从一组页面ID或数据库ID加载内容。
如果不知道页面ID或数据库ID，请不要使用此端点。
Source code in llama_index/tools/notion/base.py
54
55
56
57
58
59
60
61
62
63 | def load_data(
self, page_ids: Optional[List[str]] = None, database_id: Optional[str] = None
) -> str:
"""从一组页面ID或数据库ID加载内容。
如果不知道页面ID或数据库ID，请不要使用此端点。
"""
page_ids = page_ids or []
docs = self.reader.load_data(page_ids=page_ids, database_id=database_id)
return "\n".join([doc.get_content() for doc in docs])
|
search_data(
query: str,
direction: Optional[str] = None,
timestamp: Optional[str] = None,
value: Optional[str] = None,
property: Optional[str] = None,
page_size: int = 100,
) -> str
搜索相关页面的列表。
包含每个页面的元数据（但不包括页面内容）。
Source code in llama_index/tools/notion/base.py
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98 | def search_data(
self,
query: str,
direction: Optional[str] = None,
timestamp: Optional[str] = None,
value: Optional[str] = None,
property: Optional[str] = None,
page_size: int = 100,
) -> str:
"""搜索相关页面的列表。
包含每个页面的元数据（但不包括页面内容）。
"""
payload: Dict[str, Any] = {
"query": query,
"page_size": page_size,
}
if direction is not None or timestamp is not None:
payload["sort"] = {}
if direction is not None:
payload["sort"]["direction"] = direction
if timestamp is not None:
payload["sort"]["timestamp"] = timestamp
if value is not None or property is not None:
payload["filter"] = {}
if value is not None:
payload["filter"]["value"] = value
if property is not None:
payload["filter"]["property"] = property
response = requests.post(SEARCH_URL, json=payload, headers=self.reader.headers)
response_json = response.json()
return response_json["results"]
|