PebbloLoaderAPIWrapper#

class langchain_community.utilities.pebblo.PebbloLoaderAPIWrapper[源代码]#

基础类：BaseModel

Pebblo Loader API 的封装器。

验证环境中的API密钥。

param anonymize_snippets: bool = False#

是否对进入VectorDB的代码片段和生成的报告进行匿名化处理

param api_key: str | None [Required]#

Pebblo云的API密钥

param classifier_location: str = 'local'#

分类器的位置，本地或云端。默认为‘local’

param classifier_url: str | None [Required]#

Pebblo分类器的URL

param cloud_url: str | None [Required]#

Pebblo云的URL

build_classification_payload(app: App, docs: List[dict], loader_details: dict, source_owner: str, source_aggregate_size: int, loading_end: bool) → dict[source]#

构建文档分类的有效载荷。

Parameters:

  • app (App) – 应用程序实例。

  • docs (List[dict]) – 需要分类的文档列表。

  • loader_details (dict) – 加载器详细信息。

  • source_owner (str) – 源的所有者。

  • source_aggregate_size (int) – 源的聚合大小。

  • loading_end (bool) – 布尔值，表示加载器是否停止数据加载。

Returns:

文档分类的有效载荷。

Return type:

字典

classify_documents(docs_with_id: List[IndexedDocument], app: App, loader_details: dict, loading_end: bool = False) → dict[来源]#

将文档发送到Pebblo服务器进行分类。 然后将分类后的文档发送到Daxa云（如果存在api_key）。

Parameters:

  • docs_with_id (List[IndexedDocument]) – 需要分类的文档列表。

  • app (App) – 应用程序实例。

  • loader_details (dict) – 加载器详细信息。

  • loading_end (bool) – 布尔值，表示加载器停止数据加载。

Return type:

字典

static make_request(method: str, url: str, headers: dict, payload: dict | None = None, timeout: int = 20) → Response | None[source]#

向Pebblo API发出请求

Parameters:

  • method (str) – HTTP 方法 (GET, POST, PUT, DELETE, 等).

  • url (str) – 请求的URL。

  • headers (dict) – 请求的头部信息。

  • payload (可选[字典]) – 请求的有效载荷（用于POST、PUT等）。

  • timeout (int) – 请求的超时时间，单位为秒。

Returns:

如果请求成功，返回响应对象。

Return type:

可选[响应]

static prepare_docs_for_classification(docs_with_id: List[IndexedDocument], source_path: str, loader_details: dict) → Tuple[List[dict], int][source]#

准备分类文件。

Parameters:

  • docs_with_id (List[IndexedDocument]) – 需要分类的文档列表。

  • source_path (str) – 文档的源路径。

  • loader_details (dict) – 包含加载器信息。

Returns:

文档及其源的总大小。

Return type:

元组[列表[字典], 整数]

send_docs_to_pebblo_cloud(payload: dict) → None[source]#

将文档发送到Pebblo云。

Parameters:

payload (dict) – 包含要发送的文档的有效载荷。

Return type:

无

send_loader_discover(app: App) → None[source]#

发送应用程序发现请求到Pebblo服务器和云。

Parameters:

app (App) – 要发现的App实例。

Return type:

无

static update_doc_data(docs: List[dict], classified_docs: dict) → None[source]#

使用分类信息更新文档数据。

Parameters:

  • docs (List[dict]) – 要更新的文档数据列表。

  • classified_docs (dict) – 包含分类文档的字典。

Return type:

无