fix:修复不启动意图识别,functioncall依旧加载的bug (#691)

update:intent_llm意图识别获取新闻和天气过长,tts容易出错,暂时只支持简单简单工具
This commit is contained in:
hrz
2025-04-07 13:23:22 +08:00
committed by GitHub
parent 17bcaba026
commit d9cd9acd27
10 changed files with 228 additions and 181 deletions
Binary file not shown.

Before

Width:  |  Height:  |  Size: 396 KiB

After

Width:  |  Height:  |  Size: 579 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 114 KiB

After

Width:  |  Height:  |  Size: 161 KiB

+46 -40
View File
@@ -132,9 +132,6 @@ class ConnectionHandler:
await self.websocket.send(json.dumps(self.welcome_msg)) await self.websocket.send(json.dumps(self.welcome_msg))
# Load private configuration if device_id is provided # Load private configuration if device_id is provided
bUsePrivateConfig = self.config.get("use_private_config", False) bUsePrivateConfig = self.config.get("use_private_config", False)
self.logger.bind(tag=TAG).info(
f"bUsePrivateConfig: {bUsePrivateConfig}, device_id: {device_id}"
)
if bUsePrivateConfig and device_id: if bUsePrivateConfig and device_id:
try: try:
self.private_config = PrivateConfig( self.private_config = PrivateConfig(
@@ -220,50 +217,58 @@ class ConnectionHandler:
self.prompt = self.private_config.private_config.get("prompt", self.prompt) self.prompt = self.private_config.private_config.get("prompt", self.prompt)
self.dialogue.put(Message(role="system", content=self.prompt)) self.dialogue.put(Message(role="system", content=self.prompt))
"""加载插件"""
self.func_handler = FunctionHandler(self)
self.mcp_manager = MCPManager(self)
"""加载记忆""" """加载记忆"""
device_id = self.headers.get("device-id", None) self._initialize_memory()
self.memory.init_memory(device_id, self.llm) """加载意图识别"""
self._initialize_intent()
"""为意图识别设置LLM,优先使用专用LLM"""
# 检查是否配置了专用的意图识别LLM
intent_llm_name = self.config["Intent"]["intent_llm"]["llm"]
# 记录开始初始化意图识别LLM的时间
intent_llm_init_start = time.time()
if (
not self.use_function_call_mode
and intent_llm_name
and intent_llm_name in self.config["LLM"]
):
# 如果配置了专用LLM,则创建独立的LLM实例
from core.utils import llm as llm_utils
intent_llm_config = self.config["LLM"][intent_llm_name]
intent_llm_type = intent_llm_config.get("type", intent_llm_name)
intent_llm = llm_utils.create_instance(intent_llm_type, intent_llm_config)
self.logger.bind(tag=TAG).info(
f"为意图识别创建了专用LLM: {intent_llm_name}, 类型: {intent_llm_type}"
)
self.intent.set_llm(intent_llm)
else:
# 否则使用主LLM
self.intent.set_llm(self.llm)
# 记录意图识别LLM初始化耗时
intent_llm_init_time = time.time() - intent_llm_init_start
"""加载位置信息""" """加载位置信息"""
self.client_ip_info = get_ip_info(self.client_ip) self.client_ip_info = get_ip_info(self.client_ip)
if self.client_ip_info is not None and "city" in self.client_ip_info: if self.client_ip_info is not None and "city" in self.client_ip_info:
self.logger.bind(tag=TAG).info(f"Client ip info: {self.client_ip_info}") self.logger.bind(tag=TAG).info(f"Client ip info: {self.client_ip_info}")
self.prompt = self.prompt + f"\nuser location:{self.client_ip_info}" self.prompt = self.prompt + f"\nuser location:{self.client_ip_info}"
self.dialogue.update_system_message(self.prompt) self.dialogue.update_system_message(self.prompt)
def _initialize_memory(self):
"""初始化记忆模块"""
device_id = self.headers.get("device-id", None)
self.memory.init_memory(device_id, self.llm)
def _initialize_intent(self):
"""初始化意图识别模块"""
# 获取意图识别配置
intent_config = self.config["Intent"]
intent_type = self.config["selected_module"]["Intent"]
# 如果使用 nointent,直接返回
if intent_type == "nointent":
return
# 使用 intent_llm 模式
elif intent_type == "intent_llm":
intent_llm_name = intent_config["intent_llm"]["llm"]
if intent_llm_name and intent_llm_name in self.config["LLM"]:
# 如果配置了专用LLM,则创建独立的LLM实例
from core.utils import llm as llm_utils
intent_llm_config = self.config["LLM"][intent_llm_name]
intent_llm_type = intent_llm_config.get("type", intent_llm_name)
intent_llm = llm_utils.create_instance(
intent_llm_type, intent_llm_config
)
self.logger.bind(tag=TAG).info(
f"为意图识别创建了专用LLM: {intent_llm_name}, 类型: {intent_llm_type}"
)
self.intent.set_llm(intent_llm)
else:
# 否则使用主LLM
self.intent.set_llm(self.llm)
self.logger.bind(tag=TAG).info("使用主LLM作为意图识别模型")
"""加载插件"""
self.func_handler = FunctionHandler(self)
self.mcp_manager = MCPManager(self)
"""加载MCP工具""" """加载MCP工具"""
asyncio.run_coroutine_threadsafe( asyncio.run_coroutine_threadsafe(
self.mcp_manager.initialize_servers(), self.loop self.mcp_manager.initialize_servers(), self.loop
@@ -769,7 +774,8 @@ class ConnectionHandler:
async def close(self, ws=None): async def close(self, ws=None):
"""资源清理方法""" """资源清理方法"""
# 清理MCP资源 # 清理MCP资源
await self.mcp_manager.cleanup_all() if hasattr(self, "mcp_manager") and self.mcp_manager:
await self.mcp_manager.cleanup_all()
# 触发停止事件并清理资源 # 触发停止事件并清理资源
if self.stop_event: if self.stop_event:
@@ -69,12 +69,13 @@ async def process_intent_result(conn, intent_result, original_text):
# 检查是否有function_call # 检查是否有function_call
if "function_call" in intent_data: if "function_call" in intent_data:
# 直接从意图识别获取了function_call # 直接从意图识别获取了function_call
logger.bind(tag=TAG).info( logger.bind(tag=TAG).debug(
f"检测到function_call格式的意图结果: {intent_data['function_call']['name']}" f"检测到function_call格式的意图结果: {intent_data['function_call']['name']}"
) )
function_name = intent_data["function_call"]["name"] function_name = intent_data["function_call"]["name"]
if function_name == "continue_chat": if function_name == "continue_chat":
return False return False
function_args = None function_args = None
if "arguments" in intent_data["function_call"]: if "arguments" in intent_data["function_call"]:
function_args = intent_data["function_call"]["arguments"] function_args = intent_data["function_call"]["arguments"]
@@ -263,6 +263,8 @@ def register_device_type(descriptor):
# 用于接受前端设备推送的搜索iot描述 # 用于接受前端设备推送的搜索iot描述
async def handleIotDescriptors(conn, descriptors): async def handleIotDescriptors(conn, descriptors):
if not conn.use_function_call_mode:
return
wait_max_time = 5 wait_max_time = 5
while conn.func_handler is None or not conn.func_handler.finish_init: while conn.func_handler is None or not conn.func_handler.finish_init:
await asyncio.sleep(1) await asyncio.sleep(1)
@@ -9,20 +9,23 @@ logger = setup_logging()
class IntentProviderBase(ABC): class IntentProviderBase(ABC):
def __init__(self, config): def __init__(self, config):
self.config = config self.config = config
self.intent_options = config.get("intent_options", { self.intent_options = [
"handle_exit_intent": "结束聊天, 用户发来如再见之类的表示结束的话, 不想再进行对话的时候", {
"play_music": "播放音乐, 用户希望你可以播放音乐, 只用于播放音乐的意图", "name": "handle_exit_intent",
"get_weather": "查询天气, 用户希望查询某个地点的天气情况", "desc": "结束聊天, 用户发来如再见之类的表示结束的话, 不想再进行对话的时候",
"get_news": "查询新闻, 用户希望查询最新新闻或特定类型的新闻", },
"get_lunar": "用于获取今天的阴历/农历和黄历信息", {
"get_time": "获取今天日期或者当前时间信息", "name": "play_music",
"continue_chat": "继续聊天", "desc": "播放音乐, 用户希望你可以播放音乐, 只用于播放音乐的意图",
}) },
{"name": "get_time", "desc": "获取今天日期或者当前时间信息"},
{"name": "continue_chat", "desc": "继续聊天"},
]
def set_llm(self, llm): def set_llm(self, llm):
self.llm = llm self.llm = llm
# 获取模型名称和类型信息 # 获取模型名称和类型信息
model_name = getattr(llm, 'model_name', str(llm.__class__.__name__)) model_name = getattr(llm, "model_name", str(llm.__class__.__name__))
# 记录更详细的日志 # 记录更详细的日志
logger.bind(tag=TAG).info(f"意图识别设置LLM: {model_name}") logger.bind(tag=TAG).info(f"意图识别设置LLM: {model_name}")
@@ -35,7 +38,7 @@ class IntentProviderBase(ABC):
Returns: Returns:
返回识别出的意图,格式为: 返回识别出的意图,格式为:
- "继续聊天" - "继续聊天"
- "结束聊天" - "结束聊天"
- "播放音乐 歌名""随机播放音乐" - "播放音乐 歌名""随机播放音乐"
- "查询天气 地点名""查询天气 [当前位置]" - "查询天气 地点名""查询天气 [当前位置]"
""" """
@@ -16,5 +16,7 @@ class IntentProvider(IntentProviderBase):
Returns: Returns:
固定返回"继续聊天" 固定返回"继续聊天"
""" """
logger.bind(tag=TAG).debug("Using functionCallProvider, always returning continue chat") logger.bind(tag=TAG).debug(
return self.intent_options["continue_chat"] "Using functionCallProvider, always returning continue chat"
)
return '{"function_call": {"name": "continue_chat"}}'
@@ -27,119 +27,94 @@ class IntentProvider(IntentProviderBase):
Returns: Returns:
格式化后的系统提示词 格式化后的系统提示词
""" """
intent_list = []
prompt = ( prompt = (
"你是一个意图识别助手。请分析用户的最后一句话,判断用户意图属于以下哪一类:\n" "你是一个意图识别助手。请分析用户的最后一句话,判断用户意图属于以下哪一类:\n"
"<start>" "<start>"
f"{', '.join(intent_list)}" f"{str(self.intent_options)}"
"<end>\n" "<end>\n"
"处理步骤:" "处理步骤:"
"1. 思考意图类型,生成function_call格式" "1. 思考意图类型,生成function_call格式"
"\n\n" "\n\n"
"返回格式示例:\n" "返回格式示例:\n"
"1. 播放音乐意图: {\"function_call\": {\"name\": \"play_music\", \"arguments\": {\"song_name\": \"音乐名称\"}}}\n" '1. 播放音乐意图: {"function_call": {"name": "play_music", "arguments": {"song_name": "音乐名称"}}}\n'
"2. 查询天气意图: {\"function_call\": {\"name\": \"get_weather\", \"arguments\": {\"location\": \"地点名称\", \"lang\": \"zh_CN\"}}}\n" '2. 结束对话意图: {"function_call": {"name": "handle_exit_intent", "arguments": {"say_goodbye": "goodbye"}}}\n'
"3. 查询新闻意图: {\"function_call\": {\"name\": \"get_news\", \"arguments\": {\"category\": \"新闻类别\", \"detail\": false, \"lang\": \"zh_CN\"}}}\n" '3. 获取当天日期时间: {"function_call": {"name": "get_time"}}\n'
"4. 结束对话意图: {\"function_call\": {\"name\": \"handle_exit_intent\", \"arguments\": {\"say_goodbye\": \"goodbye\"}}}\n" '4. 继续聊天意图: {"function_call": {"name": "continue_chat"}}\n'
"5. 获取当天日期时间: {\"function_call\": {\"name\": \"get_time\"}}\n"
"6. 获取当前黄历意图: {\"function_call\": {\"name\": \"get_lunar\"}}\n"
"7. 继续聊天意图: {\"function_call\": {\"name\": \"continue_chat\"}}\n"
"\n" "\n"
"注意:\n" "注意:\n"
"- 播放音乐:无歌名时,song_name设为\"random\"\n" '- 播放音乐:无歌名时,song_name设为"random"\n'
"- 查询天气:无地点时,location设为null\n"
"- 查询新闻:无类别时,category设为null;查询详情时,detail设为true\n"
"- 如果没有明显的意图,应按照继续聊天意图处理\n" "- 如果没有明显的意图,应按照继续聊天意图处理\n"
"- 只返回纯JSON,不要任何其他内容\n" "- 只返回纯JSON,不要任何其他内容\n"
"\n" "\n"
"示例分析:\n" "示例分析:\n"
"```\n" "```\n"
"用户: 你好小智\n" "用户: 你也太搞笑了\n"
"返回: {\"function_call\": {\"name\": \"continue_chat\"}}\n" '返回: {"function_call": {"name": "continue_chat"}}\n'
"```\n"
"```\n"
"用户: 你今天怎么样?\n"
"返回: {\"function_call\": {\"name\": \"continue_chat\"}}\n"
"```\n" "```\n"
"```\n" "```\n"
"用户: 现在是几号了?现在几点了?\n" "用户: 现在是几号了?现在几点了?\n"
"返回: {\"function_call\": {\"name\": \"get_time\"}}\n" '返回: {"function_call": {"name": "get_time"}}\n'
"```\n"
"```\n"
"用户: 今天农历是多少?\n"
"返回: {\"function_call\": {\"name\": \"get_lunar\"}}\n"
"```\n" "```\n"
"```\n" "```\n"
"用户: 我们明天再聊吧\n" "用户: 我们明天再聊吧\n"
"返回: {\"function_call\": {\"name\": \"handle_exit_intent\"}}\n" '返回: {"function_call": {"name": "handle_exit_intent"}}\n'
"```\n" "```\n"
"```\n" "```\n"
"用户: 播放中秋月\n" "用户: 播放中秋月\n"
"返回: {\"function_call\": {\"name\": \"play_music\", \"arguments\": {\"song_name\": \"中秋月\"}}}\n" '返回: {"function_call": {"name": "play_music", "arguments": {"song_name": "中秋月"}}}\n'
"```\n" "```\n"
"```\n" "```\n"
"用户: 北京天气怎么样\n"
"返回: {\"function_call\": {\"name\": \"get_weather\", \"arguments\": {\"location\": \"北京\", \"lang\": \"zh_CN\"}}}\n"
"```\n"
"```\n"
"用户: 今天天气怎么样\n"
"返回: {\"function_call\": {\"name\": \"get_weather\", \"arguments\": {\"location\": null, \"lang\": \"zh_CN\"}}}\n"
"```\n"
"```\n"
"用户: 播报财经新闻\n"
"返回: {\"function_call\": {\"name\": \"get_news\", \"arguments\": {\"category\": \"财经\", \"detail\": false, \"lang\": \"zh_CN\"}}}\n"
"```\n"
"```\n"
"用户: 有什么最新新闻\n"
"返回: {\"function_call\": {\"name\": \"get_news\", \"arguments\": {\"category\": null, \"detail\": false, \"lang\": \"zh_CN\"}}}\n"
"```\n"
"```\n"
"用户: 详细介绍一下这条新闻\n"
"返回: {\"function_call\": {\"name\": \"get_news\", \"arguments\": {\"detail\": true, \"lang\": \"zh_CN\"}}}\n"
"```\n"
"可用的音乐名称:\n" "可用的音乐名称:\n"
) )
return prompt return prompt
def clean_cache(self): def clean_cache(self):
"""清理过期缓存""" """清理过期缓存"""
now = time.time() now = time.time()
# 找出过期键 # 找出过期键
expired_keys = [k for k, v in self.intent_cache.items() if now - v['timestamp'] > self.cache_expiry] expired_keys = [
k
for k, v in self.intent_cache.items()
if now - v["timestamp"] > self.cache_expiry
]
for key in expired_keys: for key in expired_keys:
del self.intent_cache[key] del self.intent_cache[key]
# 如果缓存太大,移除最旧的条目 # 如果缓存太大,移除最旧的条目
if len(self.intent_cache) > self.cache_max_size: if len(self.intent_cache) > self.cache_max_size:
# 按时间戳排序并保留最新的条目 # 按时间戳排序并保留最新的条目
sorted_items = sorted(self.intent_cache.items(), key=lambda x: x[1]['timestamp']) sorted_items = sorted(
for key, _ in sorted_items[:len(sorted_items) - self.cache_max_size]: self.intent_cache.items(), key=lambda x: x[1]["timestamp"]
)
for key, _ in sorted_items[: len(sorted_items) - self.cache_max_size]:
del self.intent_cache[key] del self.intent_cache[key]
async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str: async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str:
if not self.llm: if not self.llm:
raise ValueError("LLM provider not set") raise ValueError("LLM provider not set")
# 记录整体开始时间 # 记录整体开始时间
total_start_time = time.time() total_start_time = time.time()
# 打印使用的模型信息 # 打印使用的模型信息
model_info = getattr(self.llm, 'model_name', str(self.llm.__class__.__name__)) model_info = getattr(self.llm, "model_name", str(self.llm.__class__.__name__))
logger.bind(tag=TAG).info(f"使用意图识别模型: {model_info}") logger.bind(tag=TAG).debug(f"使用意图识别模型: {model_info}")
# 计算缓存键 # 计算缓存键
cache_key = hashlib.md5(text.encode()).hexdigest() cache_key = hashlib.md5(text.encode()).hexdigest()
# 检查缓存 # 检查缓存
if cache_key in self.intent_cache: if cache_key in self.intent_cache:
cache_entry = self.intent_cache[cache_key] cache_entry = self.intent_cache[cache_key]
# 检查缓存是否过期 # 检查缓存是否过期
if time.time() - cache_entry['timestamp'] <= self.cache_expiry: if time.time() - cache_entry["timestamp"] <= self.cache_expiry:
cache_time = time.time() - total_start_time cache_time = time.time() - total_start_time
logger.bind(tag=TAG).info(f"使用缓存的意图: {cache_key} -> {cache_entry['intent']}, 耗时: {cache_time:.4f}") logger.bind(tag=TAG).debug(
return cache_entry['intent'] f"使用缓存的意图: {cache_key} -> {cache_entry['intent']}, 耗时: {cache_time:.4f}"
)
return cache_entry["intent"]
# 清理缓存 # 清理缓存
self.clean_cache() self.clean_cache()
@@ -158,38 +133,41 @@ class IntentProvider(IntentProviderBase):
music_file_names = music_config["music_file_names"] music_file_names = music_config["music_file_names"]
prompt_music = f"{self.promot}\n<start>{music_file_names}\n<end>" prompt_music = f"{self.promot}\n<start>{music_file_names}\n<end>"
logger.bind(tag=TAG).debug(f"User prompt: {prompt_music}") logger.bind(tag=TAG).debug(f"User prompt: {prompt_music}")
# 记录预处理完成时间 # 记录预处理完成时间
preprocess_time = time.time() - total_start_time preprocess_time = time.time() - total_start_time
logger.bind(tag=TAG).debug(f"意图识别预处理耗时: {preprocess_time:.4f}") logger.bind(tag=TAG).debug(f"意图识别预处理耗时: {preprocess_time:.4f}")
# 使用LLM进行意图识别 # 使用LLM进行意图识别
llm_start_time = time.time() llm_start_time = time.time()
logger.bind(tag=TAG).info(f"开始LLM意图识别调用, 模型: {model_info}") logger.bind(tag=TAG).debug(f"开始LLM意图识别调用, 模型: {model_info}")
intent = self.llm.response_no_stream( intent = self.llm.response_no_stream(
system_prompt=prompt_music, system_prompt=prompt_music, user_prompt=user_prompt
user_prompt=user_prompt
) )
# 记录LLM调用完成时间 # 记录LLM调用完成时间
llm_time = time.time() - llm_start_time llm_time = time.time() - llm_start_time
logger.bind(tag=TAG).info(f"LLM意图识别完成, 模型: {model_info}, 调用耗时: {llm_time:.4f}") logger.bind(tag=TAG).debug(
f"LLM意图识别完成, 模型: {model_info}, 调用耗时: {llm_time:.4f}"
)
# 记录后处理开始时间 # 记录后处理开始时间
postprocess_start_time = time.time() postprocess_start_time = time.time()
# 清理和解析响应 # 清理和解析响应
intent = intent.strip() intent = intent.strip()
# 尝试提取JSON部分 # 尝试提取JSON部分
match = re.search(r'\{.*\}', intent, re.DOTALL) match = re.search(r"\{.*\}", intent, re.DOTALL)
if match: if match:
intent = match.group(0) intent = match.group(0)
# 记录总处理时间 # 记录总处理时间
total_time = time.time() - total_start_time total_time = time.time() - total_start_time
logger.bind(tag=TAG).info(f"【意图识别性能】模型: {model_info}, 总耗时: {total_time:.4f}秒, LLM调用: {llm_time:.4f}秒, 查询: '{text[:20]}...'") logger.bind(tag=TAG).debug(
f"【意图识别性能】模型: {model_info}, 总耗时: {total_time:.4f}秒, LLM调用: {llm_time:.4f}秒, 查询: '{text[:20]}...'"
)
# 尝试解析为JSON # 尝试解析为JSON
try: try:
intent_data = json.loads(intent) intent_data = json.loads(intent)
@@ -198,38 +176,42 @@ class IntentProvider(IntentProviderBase):
function_data = intent_data["function_call"] function_data = intent_data["function_call"]
function_name = function_data.get("name") function_name = function_data.get("name")
function_args = function_data.get("arguments", {}) function_args = function_data.get("arguments", {})
# 记录识别到的function call # 记录识别到的function call
logger.bind(tag=TAG).info(f"识别到function call: {function_name}, 参数: {function_args}") logger.bind(tag=TAG).info(
f"识别到function call: {function_name}, 参数: {function_args}"
)
# 添加到缓存 # 添加到缓存
self.intent_cache[cache_key] = { self.intent_cache[cache_key] = {
'intent': intent, "intent": intent,
'timestamp': time.time() "timestamp": time.time(),
} }
# 后处理时间 # 后处理时间
postprocess_time = time.time() - postprocess_start_time postprocess_time = time.time() - postprocess_start_time
logger.bind(tag=TAG).debug(f"意图后处理耗时: {postprocess_time:.4f}") logger.bind(tag=TAG).debug(f"意图后处理耗时: {postprocess_time:.4f}")
# 确保返回完全序列化的JSON字符串 # 确保返回完全序列化的JSON字符串
return intent return intent
else: else:
# 添加到缓存 # 添加到缓存
self.intent_cache[cache_key] = { self.intent_cache[cache_key] = {
'intent': intent, "intent": intent,
'timestamp': time.time() "timestamp": time.time(),
} }
# 后处理时间 # 后处理时间
postprocess_time = time.time() - postprocess_start_time postprocess_time = time.time() - postprocess_start_time
logger.bind(tag=TAG).debug(f"意图后处理耗时: {postprocess_time:.4f}") logger.bind(tag=TAG).debug(f"意图后处理耗时: {postprocess_time:.4f}")
# 返回普通意图 # 返回普通意图
return intent return intent
except json.JSONDecodeError: except json.JSONDecodeError:
# 后处理时间 # 后处理时间
postprocess_time = time.time() - postprocess_start_time postprocess_time = time.time() - postprocess_start_time
logger.bind(tag=TAG).error(f"无法解析意图JSON: {intent}, 后处理耗时: {postprocess_time:.4f}") logger.bind(tag=TAG).error(
f"无法解析意图JSON: {intent}, 后处理耗时: {postprocess_time:.4f}"
)
# 如果解析失败,默认返回继续聊天意图 # 如果解析失败,默认返回继续聊天意图
return "{\"intent\": \"继续聊天\"}" return '{"intent": "继续聊天"}'
@@ -16,5 +16,7 @@ class IntentProvider(IntentProviderBase):
Returns: Returns:
固定返回"继续聊天" 固定返回"继续聊天"
""" """
logger.bind(tag=TAG).debug("Using NoIntentProvider, always returning continue chat") logger.bind(tag=TAG).debug(
return self.intent_options["continue_chat"] "Using NoIntentProvider, always returning continue chat"
)
return '{"function_call": {"name": "continue_chat"}}'
@@ -2,15 +2,28 @@ from datetime import datetime
import cnlunar import cnlunar
from plugins_func.register import register_function, ToolType, ActionResponse, Action from plugins_func.register import register_function, ToolType, ActionResponse, Action
# 添加星期映射字典
WEEKDAY_MAP = {
"Monday": "星期一",
"Tuesday": "星期二",
"Wednesday": "星期三",
"Thursday": "星期四",
"Friday": "星期五",
"Saturday": "星期六",
"Sunday": "星期日",
}
get_time_function_desc = { get_time_function_desc = {
"type": "function", "type": "function",
"function": { "function": {
"name": "get_time", "name": "get_time",
"description": "获取今天日期或者当前时间信息", "description": "获取今天日期或者当前时间信息",
'parameters': {'type': 'object', 'properties': {}, 'required': []} "parameters": {"type": "object", "properties": {}, "required": []},
} },
} }
@register_function('get_time', get_time_function_desc, ToolType.WAIT)
@register_function("get_time", get_time_function_desc, ToolType.WAIT)
def get_time(): def get_time():
""" """
获取当前的日期时间信息 获取当前的日期时间信息
@@ -18,8 +31,10 @@ def get_time():
now = datetime.now() now = datetime.now()
current_time = now.strftime("%H:%M:%S") current_time = now.strftime("%H:%M:%S")
current_date = now.strftime("%Y-%m-%d") current_date = now.strftime("%Y-%m-%d")
current_weekday = now.strftime("%A") current_weekday = WEEKDAY_MAP[now.strftime("%A")]
response_text = f"当前日期: {current_date},当前时间: {current_time},星期: {current_weekday}" response_text = (
f"当前日期: {current_date},当前时间: {current_time} {current_weekday}"
)
return ActionResponse(Action.REQLLM, response_text, None) return ActionResponse(Action.REQLLM, response_text, None)
@@ -29,23 +44,25 @@ get_lunar_function_desc = {
"function": { "function": {
"name": "get_lunar", "name": "get_lunar",
"description": ( "description": (
"用于获取今天的阴历/农历和黄历信息。" "用于获取今天的阴历/农历和黄历信息。"
"用户可以指定查询内容,如:阴历日期、天干地支、节气、生肖、星座、八字、宜忌等。" "用户可以指定查询内容,如:阴历日期、天干地支、节气、生肖、星座、八字、宜忌等。"
"如果没有指定查询内容,则默认查询干支年和农历日期。" "如果没有指定查询内容,则默认查询干支年和农历日期。"
), ),
"parameters": { "parameters": {
"type": "object", "type": "object",
"properties": { "properties": {
"query": { "query": {
"type": "string", "type": "string",
"description": "要查询的内容,例如阴历日期、天干地支、节日、节气、生肖、星座、八字、宜忌等" "description": "要查询的内容,例如阴历日期、天干地支、节日、节气、生肖、星座、八字、宜忌等",
} }
}, },
"required": [] "required": [],
} },
} },
} }
@register_function('get_lunar', get_lunar_function_desc, ToolType.WAIT)
@register_function("get_lunar", get_lunar_function_desc, ToolType.WAIT)
def get_lunar(query=None): def get_lunar(query=None):
""" """
用于获取当前的阴历/农历,和天干地支、节气、生肖、星座、八字、宜忌等黄历信息 用于获取当前的阴历/农历,和天干地支、节气、生肖、星座、八字、宜忌等黄历信息
@@ -53,37 +70,69 @@ def get_lunar(query=None):
now = datetime.now() now = datetime.now()
current_time = now.strftime("%H:%M:%S") current_time = now.strftime("%H:%M:%S")
current_date = now.strftime("%Y-%m-%d") current_date = now.strftime("%Y-%m-%d")
current_weekday = now.strftime("%A") current_weekday = WEEKDAY_MAP[now.strftime("%A")]
# 如果 query 为 None,则使用默认文本 # 如果 query 为 None,则使用默认文本
if query is None: if query is None:
query = "默认查询干支年和农历日期" query = "默认查询干支年和农历日期"
response_text = f"根据以下信息回应用户的查询请求,并提供与{query}相关的信息:\n" response_text = f"根据以下信息回应用户的查询请求,并提供与{query}相关的信息:\n"
lunar = cnlunar.Lunar(now, godType='8char') lunar = cnlunar.Lunar(now, godType="8char")
response_text += ( response_text += (
f"当前公历日期: {current_date},当前时间: {current_time}星期: {current_weekday}\n" f"当前公历日期: {current_date},当前时间: {current_time}{current_weekday}\n"
"农历信息:\n" "农历信息:\n"
"%s%s%s\n" % (lunar.lunarYearCn, lunar.lunarMonthCn[:-1], lunar.lunarDayCn) + "%s%s%s\n" % (lunar.lunarYearCn, lunar.lunarMonthCn[:-1], lunar.lunarDayCn)
"干支: %s%s%s\n" % (lunar.year8Char, lunar.month8Char, lunar.day8Char) + + "干支: %s%s%s\n" % (lunar.year8Char, lunar.month8Char, lunar.day8Char)
"生肖: 属%s\n" % (lunar.chineseYearZodiac) + + "生肖: 属%s\n" % (lunar.chineseYearZodiac)
"八字: %s\n" % (' '.join([lunar.year8Char, lunar.month8Char, lunar.day8Char, lunar.twohour8Char])) + + "八字: %s\n"
"今日节日: %s\n" % (",".join(filter(None, (lunar.get_legalHolidays(), lunar.get_otherHolidays(), lunar.get_otherLunarHolidays())))) + % (
"今日节气: %s\n" % (lunar.todaySolarTerms) + " ".join(
"下一节气: %s %s%s%s\n" % (lunar.nextSolarTerm, lunar.nextSolarTermYear, lunar.nextSolarTermDate[0], lunar.nextSolarTermDate[1]) + [lunar.year8Char, lunar.month8Char, lunar.day8Char, lunar.twohour8Char]
"今年节气表: %s\n" % (', '.join([f"{term}({date[0]}{date[1]}日)" for term, date in lunar.thisYearSolarTermsDic.items()])) + )
"生肖冲煞: %s\n" % (lunar.chineseZodiacClash) + )
"星座: %s\n" % (lunar.starZodiac) + + "今日节日: %s\n"
"纳音: %s\n" % lunar.get_nayin() + % (
"彭祖百忌: %s\n" % (lunar.get_pengTaboo(delimit=", ")) + ",".join(
"值日: %s执位\n" % lunar.get_today12DayOfficer()[0] + filter(
"值神: %s(%s)\n" % (lunar.get_today12DayOfficer()[1], lunar.get_today12DayOfficer()[2]) + None,
"廿八宿: %s\n" % lunar.get_the28Stars() + (
"吉神方位: %s\n" % ' '.join(lunar.get_luckyGodsDirection()) + lunar.get_legalHolidays(),
"今日胎神: %s\n" % lunar.get_fetalGod() + lunar.get_otherHolidays(),
"宜: %s\n" % ''.join(lunar.goodThing[:10]) + lunar.get_otherLunarHolidays(),
"忌: %s\n" % ''.join(lunar.badThing[:10]) + ),
"(默认返回干支年和农历日期;仅在要求查询宜忌信息时才返回本日宜忌)" )
)
)
+ "今日节气: %s\n" % (lunar.todaySolarTerms)
+ "下一节气: %s %s%s%s\n"
% (
lunar.nextSolarTerm,
lunar.nextSolarTermYear,
lunar.nextSolarTermDate[0],
lunar.nextSolarTermDate[1],
)
+ "今年节气表: %s\n"
% (
", ".join(
[
f"{term}({date[0]}{date[1]}日)"
for term, date in lunar.thisYearSolarTermsDic.items()
]
)
)
+ "生肖冲煞: %s\n" % (lunar.chineseZodiacClash)
+ "星座: %s\n" % (lunar.starZodiac)
+ "纳音: %s\n" % lunar.get_nayin()
+ "彭祖百忌: %s\n" % (lunar.get_pengTaboo(delimit=", "))
+ "值日: %s执位\n" % lunar.get_today12DayOfficer()[0]
+ "值神: %s(%s)\n"
% (lunar.get_today12DayOfficer()[1], lunar.get_today12DayOfficer()[2])
+ "廿八宿: %s\n" % lunar.get_the28Stars()
+ "吉神方位: %s\n" % " ".join(lunar.get_luckyGodsDirection())
+ "今日胎神: %s\n" % lunar.get_fetalGod()
+ "宜: %s\n" % "".join(lunar.goodThing[:10])
+ "忌: %s\n" % "".join(lunar.badThing[:10])
+ "(默认返回干支年和农历日期;仅在要求查询宜忌信息时才返回本日宜忌)"
) )
return ActionResponse(Action.REQLLM, response_text, None) return ActionResponse(Action.REQLLM, response_text, None)