100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
import pandas
|
||
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
|
||
from utils.logger_handler import logger
|
||
import os
|
||
|
||
rag = RagSummarizeService()
|
||
|
||
user_ids = [str(i) for i in range(1001, 1011, 1)]
|
||
month_arr = [f"2025-{str(i).zfill(2)}" for i in range(1, 13, 1)]
|
||
external_data = {}
|
||
|
||
|
||
@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": {"特征: ..., "效率:}
|
||
}
|
||
}"""
|
||
|
||
if not external_data:
|
||
external_data_path = agent_conf['external_data_path']
|
||
external_data_path = get_abs_path(external_data_path)
|
||
|
||
if not os.path.exists(external_data_path):
|
||
raise FileExistsError(f"外部文件{external_data_path}不存在")
|
||
|
||
with open(external_data_path, 'r', encoding='utf-8') as f:
|
||
for line in f.readlines():
|
||
arr = line.strip().split(',')
|
||
user_id = arr[0].replace('"', "")
|
||
feature = arr[1].replace('"', "")
|
||
efficiency = arr[2].replace('"', "")
|
||
consumables = arr[3].replace('"', "")
|
||
comparison = arr[4].replace('"', "")
|
||
time = arr[5].replace('"', "")
|
||
|
||
if user_id not in external_data:
|
||
external_data[user_id] = {}
|
||
|
||
external_data[user_id][time] = {
|
||
"特征": feature,
|
||
"效率": efficiency,
|
||
"耗材": consumables,
|
||
"对比": comparison,
|
||
}
|
||
return external_data
|
||
|
||
|
||
@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已调用"
|
||
|
||
# if __name__ == '__main__':
|
||
# user_id = "1005"
|
||
# month = "2025-06"
|
||
# res = fetch_external_data(user_id, month)
|
||
# print(res)
|