PebbloLoaderAPIWrapper#

class langchain_community.utilities.pebblo.PebbloLoaderAPIWrapper[source]#

Bases: BaseModel

Wrapper for Pebblo Loader API.

Validate that api key in environment.

param anonymize_snippets: bool = False#

Whether to anonymize snippets going into VectorDB and the generated reports

param api_key: str | None [Required]#

API key for Pebblo Cloud

param classifier_location: str = 'local'#

Location of the classifier, local or cloud. Defaults to ‘local’

param classifier_url: str | None [Required]#

URL of the Pebblo Classifier

param cloud_url: str | None [Required]#

URL of the Pebblo Cloud

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

Build the payload for document classification.

Parameters:

  • app (App) – App instance.

  • docs (List[dict]) – List of documents to be classified.

  • loader_details (dict) – Loader details.

  • source_owner (str) – Owner of the source.

  • source_aggregate_size (int) – Aggregate size of the source.

  • loading_end (bool) – Boolean indicating the halt of data loading by loader.

Returns:

Payload for document classification.

Return type:

dict

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

Send documents to Pebblo server for classification. Then send classified documents to Daxa cloud(If api_key is present).

Parameters:

  • docs_with_id (List[IndexedDocument]) – List of documents to be classified.

  • app (App) – App instance.

  • loader_details (dict) – Loader details.

  • loading_end (bool) – Boolean, indicating the halt of data loading by loader.

Return type:

dict

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

Make a request to the Pebblo API

Parameters:

  • method (str) – HTTP method (GET, POST, PUT, DELETE, etc.).

  • url (str) – URL for the request.

  • headers (dict) – Headers for the request.

  • payload (Optional[dict]) – Payload for the request (for POST, PUT, etc.).

  • timeout (int) – Timeout for the request in seconds.

Returns:

Response object if the request is successful.

Return type:

Optional[Response]

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

Prepare documents for classification.

Parameters:

  • docs_with_id (List[IndexedDocument]) – List of documents to be classified.

  • source_path (str) – Source path of the documents.

  • loader_details (dict) – Contains loader info.

Returns:

Documents and the aggregate size of the source.

Return type:

Tuple[List[dict], int]

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

Send documents to Pebblo cloud.

Parameters:

payload (dict) – The payload containing documents to be sent.

Return type:

None

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

Send app discovery request to Pebblo server & cloud.

Parameters:

app (App) – App instance to be discovered.

Return type:

None

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

Update the document data with classified information.

Parameters:

  • docs (List[dict]) – List of document data to be updated.

  • classified_docs (dict) – The dictionary containing classified documents.

Return type:

None