96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
import json
|
|
from datetime import datetime
|
|
from typing import Dict, Any
|
|
|
|
from langchain_core.messages import HumanMessage, SystemMessage
|
|
|
|
from agent.health import tip_agent, recipe_agent, report_agent
|
|
from schema.agent import ReportRequest
|
|
|
|
memory_store: Dict[str, Any] = {
|
|
"articles": [],
|
|
"recipe": []
|
|
}
|
|
|
|
|
|
def query_tip_agent():
|
|
try:
|
|
print("开始生成今日健康资讯")
|
|
today = datetime.today().strftime("%Y-%m-%d")
|
|
|
|
resp = tip_agent.invoke({
|
|
"messages": [SystemMessage(content=f"当前系统日期：{today} \n\n"),
|
|
HumanMessage(content="请生成今日健康资讯")]
|
|
})
|
|
last_msg = resp["messages"][-1]
|
|
raw_text = last_msg.content.strip()
|
|
|
|
# 解析大模型返回的json字符串
|
|
articles = json.loads(raw_text)
|
|
# 写入内存缓存
|
|
memory_store["articles"] = articles
|
|
print("结束生成今日健康资讯")
|
|
return True
|
|
except json.JSONDecodeError as je:
|
|
print(f"[JSON解析错误] {str(je)}")
|
|
memory_store["articles"] = []
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"\n[错误]: {str(e)}")
|
|
memory_store["articles"] = []
|
|
return False
|
|
|
|
|
|
def get_latest_tip():
|
|
return memory_store["articles"]
|
|
|
|
|
|
def query_recipe_agent():
|
|
try:
|
|
print("开始生成今日养生食谱")
|
|
today = datetime.today().strftime("%Y-%m-%d")
|
|
|
|
resp = recipe_agent.invoke({
|
|
"messages": [SystemMessage(content=f"当前系统日期：{today} \n\n"),
|
|
HumanMessage(content="请生成今日养生食谱")]
|
|
})
|
|
last_msg = resp["messages"][-1]
|
|
raw_text = last_msg.content.strip()
|
|
|
|
# 解析大模型返回的json字符串
|
|
recipes = json.loads(raw_text)
|
|
# 写入内存缓存
|
|
memory_store["recipe"] = recipes
|
|
print("结束生成今日养生食谱")
|
|
return True
|
|
except json.JSONDecodeError as je:
|
|
print(f"[JSON解析错误] {str(je)}")
|
|
memory_store["recipe"] = []
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"\n[错误]: {str(e)}")
|
|
memory_store["recipe"] = []
|
|
return False
|
|
|
|
|
|
def get_latest_recipe():
|
|
return memory_store["recipe"]
|
|
|
|
|
|
def query_report_agent(req: ReportRequest):
|
|
try:
|
|
print("开始生成报告")
|
|
today = datetime.today().strftime("%Y-%m-%d")
|
|
|
|
resp = report_agent.invoke({
|
|
"messages": [SystemMessage(content=f"当前系统日期：{today} \n\n"),
|
|
HumanMessage(content=f"用户数据：${req.query}")]
|
|
})
|
|
print("结束生成报告")
|
|
last_msg = resp["messages"][-1]
|
|
return last_msg.content.strip()
|
|
except Exception as e:
|
|
print(f"\n[生成报告错误]: {str(e)}")
|
|
return False |