File size: 4,028 Bytes
2364c5e | 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 | import os
from utils.logger_handler import logger
from langchain_core.tools import tool
from rag.rag_service import RagSummarizeService
import random
from utils.config_handler import agent_conf
from utils.path_tool import get_abs_path
user_ids = ["1001", "1002", "1003", "1004", "1005", "1006", "1007", "1008", "1009", "1010",]
month_arr = ["2025-01", "2025-02", "2025-03", "2025-04", "2025-05", "2025-06",
"2025-07", "2025-08", "2025-09", "2025-10", "2025-11", "2025-12", ]
external_data = {}
rag = RagSummarizeService()
@tool(description="从向量存储中检索参考资料")
def rag_summarize(query: str) -> str:
return rag.rag_summarize(query)
@tool(description="获取指定城市的天气，以消息字符串的形式返回")
def get_weather(city: str) -> str:
return f"城市{city}天气为晴天，气温26摄氏度，空气湿度50%，南风1级，AQI21，最近6小时降雨概率极低"
@tool(description="获取用户所在城市的名称，以纯字符串形式返回")
def get_user_location() -> str:
return random.choice(["深圳", "合肥", "杭州"])
@tool(description="获取用户的ID，以纯字符串形式返回")
def get_user_id() -> str:
return random.choice(user_ids)
@tool(description="获取当前月份，以纯字符串形式返回")
def get_current_month() -> str:
return random.choice(month_arr)
def generate_external_data():
"""
{
"user_id": {
"month" : {"特征": xxx, "效率": xxx, ...}
"month" : {"特征": xxx, "效率": xxx, ...}
"month" : {"特征": xxx, "效率": xxx, ...}
...
},
"user_id": {
"month" : {"特征": xxx, "效率": xxx, ...}
"month" : {"特征": xxx, "效率": xxx, ...}
"month" : {"特征": xxx, "效率": xxx, ...}
...
},
"user_id": {
"month" : {"特征": xxx, "效率": xxx, ...}
"month" : {"特征": xxx, "效率": xxx, ...}
"month" : {"特征": xxx, "效率": xxx, ...}
...
},
...
}
:return:
"""
if not external_data:
external_data_path = get_abs_path(agent_conf["external_data_path"])
if not os.path.exists(external_data_path):
raise FileNotFoundError(f"外部数据文件{external_data_path}不存在")
with open(external_data_path, "r", encoding="utf-8") as f:
for line in f.readlines()[1:]:
arr: list[str] = line.strip().split(",")
user_id: str = arr[0].replace('"', "")
feature: str = arr[1].replace('"', "")
efficiency: str = arr[2].replace('"', "")
consumables: str = arr[3].replace('"', "")
comparison: str = arr[4].replace('"', "")
time: str = arr[5].replace('"', "")
if user_id not in external_data:
external_data[user_id] = {}
external_data[user_id][time] = {
"特征": feature,
"效率": efficiency,
"耗材": consumables,
"对比": comparison,
}
@tool(description="从外部系统中获取指定用户在指定月份的使用记录，以纯字符串形式返回， 如果未检索到返回空字符串")
def fetch_external_data(user_id: str, month: str) -> str:
generate_external_data()
try:
return external_data[user_id][month]
except KeyError:
logger.warning(f"[fetch_external_data]未能检索到用户：{user_id}在{month}的使用记录数据")
return ""
@tool(description="无入参，无返回值，调用后触发中间件自动为报告生成的场景动态注入上下文信息，为后续提示词切换提供上下文信息")
def fill_context_for_report():
return "fill_context_for_report已调用"
|