Source code for langchain_community.utilities.golden_query

"""调用Golden的工具。"""
import json
from typing import Dict, Optional

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

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


class GoldenQueryAPIWrapper(BaseModel):
    """黄金的包装器。

    使用文档：

    1. 访问 https://golden.com 并注册账户
    2. 从 https://golden.com/settings/api 获取你的API密钥
    3. 将你的API密钥保存到GOLDEN_API_KEY环境变量中

    """

    golden_api_key: Optional[str] = None

    class Config:
        """此pydantic对象的配置。"""

        extra = Extra.forbid

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证环境中是否存在API密钥和Python包。"""
        golden_api_key = get_from_dict_or_env(
            values, "golden_api_key", "GOLDEN_API_KEY"
        )
        values["golden_api_key"] = golden_api_key

        return values

    def run(self, query: str) -> str:
        """通过Golden Query API运行查询并返回JSON原始结果。"""

        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