PebbloRetrievalAPIWrapper#

class langchain_community.chains.pebblo_retrieval.utilities.PebbloRetrievalAPIWrapper[source]#

Bases: BaseModel

Wrapper for Pebblo Retrieval API.

Validate that api key in environment.

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

async acheck_prompt_validity(question: str) → Tuple[bool, Dict[str, Any]][source]#

Check the validity of the given prompt using a remote classification service.

This method sends a prompt to a remote classifier service and return entities present in prompt or not.

Parameters:

question (str) – The prompt question to be validated.

Returns:

True if the prompt is valid (does not contain deny list entities), False otherwise. dict: The entities present in the prompt

Return type:

bool

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

Make a async request to the Pebblo server/cloud 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 json if the request is successful.

Return type:

Any

async asend_prompt(app_name: str, retriever: VectorStoreRetriever, question: str, answer: str, auth_context: AuthContext | None, docs: List[Document], prompt_entities: Dict[str, Any], prompt_time: str, prompt_gov_enabled: bool = False) → None[source]#

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

Parameters:

  • app_name (str) – Name of the app.

  • retriever (VectorStoreRetriever) – Retriever instance.

  • question (str) – Question asked in the prompt.

  • answer (str) – Answer generated by the model.

  • auth_context (Optional[AuthContext]) – Authentication context.

  • docs (List[Document]) – List of documents retrieved.

  • prompt_entities (Dict[str, Any]) – Entities present in the prompt.

  • prompt_time (str) – Time when the prompt was generated.

  • prompt_gov_enabled (bool) – Whether prompt governance is enabled.

Return type:

None

build_prompt_qa_payload(app_name: str, retriever: VectorStoreRetriever, question: str, answer: str, auth_context: AuthContext | None, docs: List[Document], prompt_entities: Dict[str, Any], prompt_time: str, prompt_gov_enabled: bool = False) → dict[source]#

Build the QA payload for the prompt.

Args:

app_name (str): Name of the app. retriever (VectorStoreRetriever): Retriever instance. question (str): Question asked in the prompt. answer (str): Answer generated by the model. auth_context (Optional[AuthContext]): Authentication context. docs (List[Document]): List of documents retrieved. prompt_entities (Dict[str, Any]): Entities present in the prompt. prompt_time (str): Time when the prompt was generated. prompt_gov_enabled (bool): Whether prompt governance is enabled.

Returns:

The QA payload for the prompt.

Return type:

dict

Parameters:

  • app_name (str)

  • retriever (VectorStoreRetriever)

  • question (str)

  • answer (str)

  • auth_context (AuthContext | None)

  • docs (List[Document])

  • prompt_entities (Dict[str, Any])

  • prompt_time (str)

  • prompt_gov_enabled (bool)

check_prompt_validity(question: str) → Tuple[bool, Dict[str, Any]][source]#

Check the validity of the given prompt using a remote classification service.

This method sends a prompt to a remote classifier service and return entities present in prompt or not.

Parameters:

question (str) – The prompt question to be validated.

Returns:

True if the prompt is valid (does not contain deny list entities), False otherwise. dict: The entities present in the prompt

Return type:

bool

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 server/cloud 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]

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

Send app discovery request to Pebblo server & cloud.

Parameters:

app (App) – App instance to be discovered.

Return type:

None

send_prompt(app_name: str, retriever: VectorStoreRetriever, question: str, answer: str, auth_context: AuthContext | None, docs: List[Document], prompt_entities: Dict[str, Any], prompt_time: str, prompt_gov_enabled: bool = False) → None[source]#

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

Parameters:

  • app_name (str) – Name of the app.

  • retriever (VectorStoreRetriever) – Retriever instance.

  • question (str) – Question asked in the prompt.

  • answer (str) – Answer generated by the model.

  • auth_context (Optional[AuthContext]) – Authentication context.

  • docs (List[Document]) – List of documents retrieved.

  • prompt_entities (Dict[str, Any]) – Entities present in the prompt.

  • prompt_time (str) – Time when the prompt was generated.

  • prompt_gov_enabled (bool) – Whether prompt governance is enabled.

Return type:

None

static update_cloud_payload(payload: dict, pebblo_resp: dict | None) → None[source]#

Update the payload with response, prompt and context from Pebblo response.

Parameters:

  • payload (dict) – Payload to be updated.

  • pebblo_resp (Optional[dict]) – Response from Pebblo server.

Return type:

None