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

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]

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

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

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