PebbloRetrievalAPIWrapper

class langchain_community.chains.pebblo_retrieval.utilities.PebbloRetrievalAPIWrapper

Bases: BaseModel

Wrapper for Pebblo Retrieval API.

Validate that api key in environment.

param api_key: str | None = None

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 = None

URL of the Pebblo Classifier

param cloud_url: str | None = None

URL of the Pebblo Cloud

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

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

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

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

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]]

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

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

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

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

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