Source code for langchain_community.utilities.golden_query

"""Util that calls Golden."""

import json
from typing import Dict, Optional

import requests
from langchain_core.pydantic_v1 import BaseModel, root_validator
from langchain_core.utils import get_from_dict_or_env

GOLDEN_BASE_URL = "https://golden.com"
GOLDEN_TIMEOUT = 5000


[docs]class GoldenQueryAPIWrapper(BaseModel): """Wrapper for Golden. Docs for using: 1. Go to https://golden.com and sign up for an account 2. Get your API Key from https://golden.com/settings/api 3. Save your API Key into GOLDEN_API_KEY env variable """ golden_api_key: Optional[str] = None class Config: extra = "forbid" @root_validator(pre=True) def validate_environment(cls, values: Dict) -> Dict: """Validate that api key and python package exists in environment.""" golden_api_key = get_from_dict_or_env( values, "golden_api_key", "GOLDEN_API_KEY" ) values["golden_api_key"] = golden_api_key return values
[docs] def run(self, query: str) -> str: """Run query through Golden Query API and return the JSON raw result.""" headers = {"apikey": self.golden_api_key or ""} response = requests.post( f"{GOLDEN_BASE_URL}/api/v2/public/queries/", json={"prompt": query}, headers=headers, timeout=GOLDEN_TIMEOUT, ) if response.status_code != 201: return response.text content = json.loads(response.content) query_id = content["id"] response = requests.get( ( f"{GOLDEN_BASE_URL}/api/v2/public/queries/{query_id}/results/" "?pageSize=10" ), headers=headers, timeout=GOLDEN_TIMEOUT, ) return response.text