Spaces:
Runtime error
Runtime error
File size: 5,542 Bytes
870e141 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
from .ToolComponent import ToolComponent
import json
from utils import flatten_dict,get_embedding,matching_category,search_with_api,limit_keys,limit_values
import os
class CategoryRequirementsComponent(ToolComponent):
def __init__(self, information_path):
super().__init__()
self.information_dataset = []
self.leaf_name = []
for toy_path in information_path:
with open(toy_path, encoding="utf-8") as json_file:
data = json.load(json_file)
for d in data:
if "/" in d["cat_leaf_name"]:
leaf_names = d["cat_leaf_name"].split("/") + [d["cat_leaf_name"]]
else:
leaf_names = [d["cat_leaf_name"]]
for name in leaf_names:
self.leaf_name.append(name)
new_d = d.copy()
new_d["cat_leaf_name"] = name
new_d["information"] = flatten_dict(new_d["information"])
self.information_dataset.append(new_d)
self.target_embbeding = get_embedding(
self.leaf_name
)
def search_information(self, category, information_dataset):
knowledge = {}
for d in information_dataset:
if category == d["cat_leaf_name"]:
knowledge = d["information"]
knowledge = {
key: value
for key, value in knowledge.items()
if (value and key != "相关分类")
}
break
return knowledge
def func(self, agent):
prompt = ""
messages = agent.long_term_memory
outputdict = {}
functions = [
{
"name": "search_information",
"description": "根据用户所需要购买商品的种类跟用户的需求去寻找用户所需要的商品",
"parameters": {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "用户现在所需要的商品类别，比如纸尿布，笔记本电脑等，注意，只能有一个",
},
"requirements": {
"type": "string",
"description": "用户现在的需求，比如说便宜，安踏品牌等等，可以有多个需求，中间以“ ”分隔",
},
},
"required": ["category", "requirements"],
},
}
]
response = agent.LLM.get_response(
messages,
None,
None,
functions=functions,
stream=False,
function_call={"name": "search_information"},
)
response_message = json.loads(response["function_call"]["arguments"])
category = (
response_message["category"] if response_message["category"] else None
)
requirements = (
response_message["requirements"]
if response_message["requirements"]
else category
)
if not (category or requirements):
return {}
topk_result = matching_category(
category, self.leaf_name, None, self.target_embbeding, top_k=3
)
top1_score = topk_result[1][0]
request_items, top_category = search_with_api(requirements, category)
MIN_CATEGORY_SIM = eval(os.environ["MIN_CATEGORY_SIM"]
) if "MIN_CATEGORY_SIM" in os.environ else 0.7
if top1_score > MIN_CATEGORY_SIM:
agent.environment.shared_memory["category"] = topk_result[0][0]
category = topk_result[0][0]
information = self.search_information(
topk_result[0][0], self.information_dataset
)
information = limit_keys(information, 3)
information = limit_values(information, 2)
prompt += f"""你需要知道的是：用户目前选择的商品是{category}，该商品信息为{information}。你需要根据这些商品信息来详细介绍商品，比如详细介绍商品有哪些品牌，有哪些分类等等，并且询问用户是否有更多的需求。"""
if category in top_category:
top_category.remove(category)
recommend = "\n经过搜索后，推荐商品如下：\n"
prompt += "筛选出的商品如下：\n"
for i, request_item in enumerate(request_items):
itemTitle = request_item["itemTitle"]
itemPrice = request_item["itemPrice"]
itemPicUrl = request_item["itemPicUrl"]
recommend += f"[{i}.商品名称：{itemTitle},商品价格:{float(itemPrice)/100}]({itemPicUrl})\n"
prompt += f"[{i}.商品名称：{itemTitle},商品价格:{float(itemPrice)/100}]\n"
outputdict["recommend"] = recommend
print(recommend)
else:
prompt += f"""你需要知道的是：用户目前选择的商品是{category}，而我们店里没有这类商品，但是我们店里有一些近似商品，如{top_category},{topk_result[0][0]}，你需要对这些近似商品进行介绍，并引导用户购买"""
outputdict["prompt"] = prompt
return outputdict
|