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