Source code for langchain_community.utilities.awslambda

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

from langchain_core.pydantic_v1 import BaseModel, Extra, root_validator


class LambdaWrapper(BaseModel):
    """AWS Lambda SDK的包装器。
    要使用，您应该已安装``boto3``包
    并且从AWS控制台或CLI构建了lambda函数。使用``aws configure``设置您的AWS凭据

    示例:
        .. code-block:: bash

            pip install boto3

            aws configure"""

    lambda_client: Any  #: :meta private:
    """配置好的boto3客户端"""
    function_name: Optional[str] = None
    """您的lambda函数的名称"""
    awslambda_tool_name: Optional[str] = None
    """如果作为工具传递给代理，工具名称"""
    awslambda_tool_description: Optional[str] = None
    """如果将其作为工具传递给代理商，则描述"""

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

        extra = Extra.forbid

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证Python包是否存在于环境中。"""

        try:
            import boto3

        except ImportError:
            raise ImportError(
                "boto3 is not installed. Please install it with `pip install boto3`"
            )

        values["lambda_client"] = boto3.client("lambda")
        values["function_name"] = values["function_name"]

        return values

    def run(self, query: str) -> str:
        """调用lambda函数并返回结果。

参数：
    query: 作为JSON对象的“body”传递给lambda函数的输入。
"""  # noqa: E501
        res = self.lambda_client.invoke(
            FunctionName=self.function_name,
            InvocationType="RequestResponse",
            Payload=json.dumps({"body": query}),
        )

        try:
            payload_stream = res["Payload"]
            payload_string = payload_stream.read().decode("utf-8")
            answer = json.loads(payload_string)["body"]

        except StopIteration:
            return "Failed to parse response from Lambda"

        if answer is None or answer == "":
            # We don't want to return the assumption alone if answer is empty
            return "Request failed."
        else:
            return f"Result: {answer}"