|
|
#!/usr/bin/env python
|
|
|
# -*- coding: utf-8 -*-
|
|
|
# @Time : 2025/4/12 14:55
|
|
|
# @Author : old-tom
|
|
|
# @File : tool_impl
|
|
|
# @Project : sttFunctionCallBackend
|
|
|
# @Desc : 工具定义
|
|
|
|
|
|
from typing import Annotated
|
|
|
from langchain_core.tools import tool
|
|
|
from vector_db import query_vector_db
|
|
|
from log_conf import log
|
|
|
from llmagent.llm_config import base_conf
|
|
|
|
|
|
|
|
|
@tool("play_video", description="播放、查看、打开实时视频")
|
|
|
def play_video(camera_name: Annotated[str, "相机名称,例如：南卡口1号相机"]) -> str:
|
|
|
camera_info = query_camera_from_db(camera_name)
|
|
|
log.info('【function】play_video 输入 [{}],向量库返回{}', camera_name, camera_info)
|
|
|
if camera_info:
|
|
|
if len(camera_info) > 1:
|
|
|
hit_camera_names = [x['carme_name'] for x in camera_info]
|
|
|
return f"【tool_back】找到以下相机，请选择一个：{hit_camera_names}"
|
|
|
else:
|
|
|
# TODO 调用业务系统
|
|
|
return f"【tool_back】正在打开{camera_name},请等待操作完成"
|
|
|
else:
|
|
|
return "【tool_back】未找到该相机,请尝试其他名称"
|
|
|
|
|
|
|
|
|
@tool("split_screen", description="切换分屏")
|
|
|
def split_screen(split_n: Annotated[int, "要切换的分屏数量,整数并且大于0，例如：1分屏、2分屏。如果是中文,例如：壹、贰、参需要转为数字形式"]) -> str:
|
|
|
return f"【tool_back】正在切换到{split_n}分屏,请等待操作完成"
|
|
|
|
|
|
|
|
|
@tool("play_video_record", description="播放、打开录像")
|
|
|
def play_video_record(camera_name: Annotated[str, "相机名称,例如：南卡口1号相机"],
|
|
|
start_time: Annotated[str, "录像开始时间,格式为yyyy-MM-dd hh:mm:ss，例 2025-03-16 01:00:00"],
|
|
|
end_time: Annotated[str, "录像结束时间,格式为yyyy-MM-dd hh:mm:ss，例 2025-03-16 02:09:31"]) -> str:
|
|
|
log.info('【function】play_video_record 输入 [{},{},{}]', camera_name, start_time, end_time)
|
|
|
return f"【tool_back】正在打开{camera_name}的录像,请等待操作完成"
|
|
|
|
|
|
|
|
|
@tool("switch_page", description="打开、跳转页面")
|
|
|
def switch_page(page_name: Annotated[str, "页面中文名称或者缩写，例：人员核查、系统日志、设备管理、首页"]) -> str:
|
|
|
log.info('【function】switch_page 输入 [{}]', page_name)
|
|
|
return f"【tool_back】正在打开{page_name},请等待操作完成"
|
|
|
|
|
|
|
|
|
@tool("zoom_in", description="放大电子地图")
|
|
|
def zoom_in(level_n: Annotated[int, "放大等级,整数并且大于0小于5，例如：放大1级、放大2级"]) -> str:
|
|
|
log.info('【function】zoom_in 输入 [{}]', level_n)
|
|
|
return f"【tool_back】正在放大电子地图至{level_n}级,请等待操作完成"
|
|
|
|
|
|
|
|
|
@tool("view_flight_details", description="查询指定机场指定航班及时间的出入境人员明细")
|
|
|
def view_flight_details(
|
|
|
airport_name: Annotated[str, "机场名称，简体中文，可以是缩写，例如：成都天府机场、天府机场、长水机场、上海浦东机场"],
|
|
|
flight_code: Annotated[
|
|
|
str, "航班编号,由字母+数字组成的完整编号,若编号包含多余字符（如标点符号），需过滤后保留有效部分"],
|
|
|
flight_date: Annotated[str, "提取完整日期（年月日），自动补零至标准格式 yyyy-MM-dd, 例：2025-03-16"],
|
|
|
ie_type: Annotated[str, "出入境类型，仅识别'入境'或'出境'两种类型"]) -> str:
|
|
|
log.info('【function】view_flight_details 输入 [{},{},{},{}]', airport_name, flight_code, flight_date, ie_type)
|
|
|
return f"【tool_back】{airport_name}航班号{flight_code}{flight_date}{ie_type}数据,共100人乘机,起飞准时,晚点降落"
|
|
|
|
|
|
|
|
|
def query_camera_from_db(camera_name: str, top_n: int = 3) -> str:
|
|
|
"""
|
|
|
相机名称查询向量库,根据相似度阈值取top_one或者top_n
|
|
|
:param camera_name: 相机名称
|
|
|
:param top_n: 返回前N个
|
|
|
:return:
|
|
|
"""
|
|
|
rt = query_vector_db(camera_name)
|
|
|
if rt:
|
|
|
log.info('【function】相机相似度检索查询[{}],返回 {}', camera_name, rt)
|
|
|
# 判断相似度最高的相机是否超过阈值
|
|
|
top_one = rt['hits'][0]
|
|
|
# 相似度评分
|
|
|
score = top_one['_score']
|
|
|
if score > base_conf.similarity_threshold:
|
|
|
return rt['hits'][0:1]
|
|
|
else:
|
|
|
return rt['hits'][0:top_n]
|
