from __future__ import annotations
from typing import Any, Dict, List, Mapping, Optional, cast
from langchain.chains import LLMChain
from langchain.chains.base import Chain
from langchain.schema.language_model import BaseLanguageModel
from langchain_core.callbacks.manager import (
CallbackManagerForChainRun,
)
from langchain_core.prompts.prompt import PromptTemplate
from langchain_experimental.recommenders.amazon_personalize import AmazonPersonalize
SUMMARIZE_PROMPT_QUERY = """
Summarize the recommended items for a user from the items list in tag <result> below.
Make correlation into the items in the list and provide a summary.
<result>
{result}
</result>
"""
SUMMARIZE_PROMPT = PromptTemplate(
input_variables=["result"], template=SUMMARIZE_PROMPT_QUERY
)
INTERMEDIATE_STEPS_KEY = "intermediate_steps"
# Input Key Names to be used
USER_ID_INPUT_KEY = "user_id"
ITEM_ID_INPUT_KEY = "item_id"
INPUT_LIST_INPUT_KEY = "input_list"
FILTER_ARN_INPUT_KEY = "filter_arn"
FILTER_VALUES_INPUT_KEY = "filter_values"
CONTEXT_INPUT_KEY = "context"
PROMOTIONS_INPUT_KEY = "promotions"
METADATA_COLUMNS_INPUT_KEY = "metadata_columns"
RESULT_OUTPUT_KEY = "result"
[docs]class AmazonPersonalizeChain(Chain):
"""Amazon Personalize的推荐检索和总结链。
只有在return_direct=True时才返回推荐结果。
它还可以用于顺序链，用于处理Amazon Personalize的输出。
示例：
.. code-block:: python
链 = PersonalizeChain.from_llm(llm=agent_llm, client=personalize_lg,
return_direct=True)
响应 = chain.run({'user_id':'1'})
响应 = chain.run({'user_id':'1', 'item_id':'234'})
"""
client: AmazonPersonalize
summarization_chain: LLMChain
return_direct: bool = False
return_intermediate_steps: bool = False
is_ranking_recipe: bool = False
@property
def input_keys(self) -> List[str]:
"""这将返回一个空列表，因为没有必需的可选输入键。
:元数据 私有:
"""
return []
@property
def output_keys(self) -> List[str]:
"""始终返回结果键。
:元数据 私有:
"""
return [RESULT_OUTPUT_KEY]
[docs] @classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
client: AmazonPersonalize,
prompt_template: PromptTemplate = SUMMARIZE_PROMPT,
is_ranking_recipe: bool = False,
**kwargs: Any,
) -> AmazonPersonalizeChain:
"""使用LLMAgent、Personalize Client和要使用的Prompts初始化Personalize Chain
参数:
llm: BaseLanguageModel: 要在Chain中使用的LLM
client: AmazonPersonalize: 用于支持调用AmazonPersonalize的客户端
prompt_template: PromptTemplate: 可以使用来自Amazon Personalize的输出来调用的提示模板
is_ranking_recipe: bool: 默认值: False: 指定训练配方是否为USER_PERSONALIZED_RANKING
示例:
.. code-block:: python
chain = PersonalizeChain.from_llm(llm=agent_llm,
client=personalize_lg, return_direct=True)
response = chain.run({'user_id':'1'})
response = chain.run({'user_id':'1', 'item_id':'234'})
RANDOM_PROMPT_QUERY=" Summarize recommendations in {result}"
chain = PersonalizeChain.from_llm(llm=agent_llm,
client=personalize_lg, prompt_template=PROMPT_TEMPLATE)
"""
summarization_chain = LLMChain(llm=llm, prompt=prompt_template)
return cls(
summarization_chain=summarization_chain,
client=client,
is_ranking_recipe=is_ranking_recipe,
**kwargs,
)
def _call(
self,
inputs: Mapping[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""通过调用Amazon Personalize检索推荐内容，并使用来自Amazon Personalize的输出调用LLM（默认/重写）提示模板。
参数：
inputs: Mapping [str, Any]：在映射中提供输入标识符。
例如 - {'user_id'，'1'}或{'user_id'：'1'，'item_id'：'123'}。您还可以将filter_arn，filter_values作为输入传递。
"""
_run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
callbacks = _run_manager.get_child()
user_id = inputs.get(USER_ID_INPUT_KEY)
item_id = inputs.get(ITEM_ID_INPUT_KEY)
input_list = inputs.get(INPUT_LIST_INPUT_KEY)
filter_arn = inputs.get(FILTER_ARN_INPUT_KEY)
filter_values = inputs.get(FILTER_VALUES_INPUT_KEY)
promotions = inputs.get(PROMOTIONS_INPUT_KEY)
context = inputs.get(CONTEXT_INPUT_KEY)
metadata_columns = inputs.get(METADATA_COLUMNS_INPUT_KEY)
intermediate_steps: List = []
intermediate_steps.append({"Calling Amazon Personalize"})
if self.is_ranking_recipe:
response = self.client.get_personalized_ranking(
user_id=str(user_id),
input_list=cast(List[str], input_list),
filter_arn=filter_arn,
filter_values=filter_values,
context=context,
metadata_columns=metadata_columns,
)
else:
response = self.client.get_recommendations(
user_id=user_id,
item_id=item_id,
filter_arn=filter_arn,
filter_values=filter_values,
context=context,
promotions=promotions,
metadata_columns=metadata_columns,
)
_run_manager.on_text("Call to Amazon Personalize complete \n")
if self.return_direct:
final_result = response
else:
result = self.summarization_chain(
{RESULT_OUTPUT_KEY: response}, callbacks=callbacks
)
final_result = result[self.summarization_chain.output_key]
intermediate_steps.append({"context": response})
chain_result: Dict[str, Any] = {RESULT_OUTPUT_KEY: final_result}
if self.return_intermediate_steps:
chain_result[INTERMEDIATE_STEPS_KEY] = intermediate_steps
return chain_result
@property
def _chain_type(self) -> str:
return "amazon_personalize_chain"