497 lines
25 KiB
Python
497 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
结构化查询工具层：供大模型函数调用（tools）与前端 REST API 共用。
|
||
所有函数返回 JSON 友好的 dict/list，并带 _source 来源标记。
|
||
新增数据域（如 CBA、足球）时：SQL 按 league/sport 过滤即可复用全部函数。
|
||
"""
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
|
||
from db import query, query_one, fuzzy
|
||
import vector_store
|
||
|
||
MAX_SHOW = 6 # 工具默认返回条数上限
|
||
|
||
# 绰号 → 库内正式名（查询时自动展开）
|
||
ALIASES = {
|
||
"字母哥": "扬尼斯·阿德托昆博", "希腊怪兽": "扬尼斯·阿德托昆博",
|
||
"老詹": "勒布朗·詹姆斯", "詹皇": "勒布朗·詹姆斯", "皇帝": "勒布朗·詹姆斯",
|
||
"浓眉": "安东尼·戴维斯", "SGA": "谢伊·吉尔杰斯-亚历山大",
|
||
"华子": "安东尼·爱德华兹", "文班": "维克托·文班亚马", "约老师": "尼古拉·约基奇",
|
||
"大胡子": "詹姆斯·哈登", "死神": "凯文·杜兰特", "大帝": "乔尔·恩比德",
|
||
"追梦": "德雷蒙德·格林", "獭兔": "杰森·塔图姆", "东子": "卢卡·东契奇",
|
||
"泡椒": "保罗·乔治", "小卡": "科怀·伦纳德", "卡皇": "亚历克斯·卡鲁索",
|
||
"切特": "切特·霍姆格伦", "杰威": "杰伦·威廉姆斯",
|
||
}
|
||
|
||
|
||
def expand_aliases(q):
|
||
"""把绰号替换为正式名（长匹配优先）"""
|
||
for nick, real in sorted(ALIASES.items(), key=lambda x: -len(x[0])):
|
||
if nick in q:
|
||
q = q.replace(nick, real)
|
||
return q
|
||
|
||
|
||
def _split_terms(q):
|
||
"""把'约基奇和字母哥'/'湖人、勇士'拆成多个关键词（绰号已展开，仅去标点）"""
|
||
q = expand_aliases(q)
|
||
terms = [t.strip() for t in re.split(r"[和与跟、,，;；vs VS 及\s]", q) if t.strip()]
|
||
terms = [re.sub(r"^[\s\-—:：]+|[\s\-—:：]+$", "", t) for t in terms]
|
||
seen, out = set(), []
|
||
for t in terms:
|
||
if t and t not in seen:
|
||
seen.add(t)
|
||
out.append(t)
|
||
return out
|
||
|
||
|
||
def _match_by_truncation(base_sql, term, args_factory, max_drop=10):
|
||
"""渐进截断匹配：整词无结果时逐个丢弃尾部字符重试（处理'字母哥谁得分多'类问句尾巴）
|
||
base_sql 需含 ? 占位；args_factory(like) 生成查询参数"""
|
||
t = term
|
||
for _ in range(max_drop + 1):
|
||
if len(t) < 2:
|
||
break
|
||
rows = query(base_sql, args_factory(f"%{fuzzy(t)}%"))
|
||
if rows:
|
||
return rows
|
||
t = t[:-1]
|
||
return []
|
||
|
||
|
||
def _fmt_team(t):
|
||
return {"id": t["id"], "name": t["name"], "name_en": t["name_en"], "code": t["code"],
|
||
"city": t["city"], "arena": t["arena"], "founded": t["founded"],
|
||
"champion_count": t["champion_count"], "head_coach": t["head_coach"], "intro": t["intro"]}
|
||
|
||
|
||
def _fmt_player(p):
|
||
return {"id": p["id"], "name": p["name"], "name_en": p["name_en"], "team": p.get("team_name"),
|
||
"team_id": p.get("team_id"),
|
||
"position": p["position"], "number": p["number"], "height_cm": p["height_cm"],
|
||
"weight_kg": p["weight_kg"], "country": p["country"], "draft": f"{p['draft_year']}年 第{p['draft_pick']}顺位" if p["draft_year"] else "落选秀",
|
||
"salary_m": p["salary_m"], "season": {"pts": p["season_pts"], "reb": p["season_reb"],
|
||
"ast": p["season_ast"], "stl": p["season_stl"], "blk": p["season_blk"], "min": p["season_min"]},
|
||
"career": {"pts": p["career_pts"], "reb": p["career_reb"], "ast": p["career_ast"], "games": p["career_games"]},
|
||
"awards": p["awards"], "bio": p["bio"]}
|
||
|
||
|
||
def _fmt_game(g, with_team=True):
|
||
d = {"id": g["id"], "round_name": g["round_name"], "game_time": g["game_time"],
|
||
"status": g["status"], "venue": g["venue"], "broadcast": g["broadcast"],
|
||
"home_score": g["home_score"], "away_score": g["away_score"]}
|
||
if with_team:
|
||
d["home_team"] = g["home_name"] or g["home_code"]
|
||
d["away_team"] = g["away_name"] or g["away_code"]
|
||
return d
|
||
|
||
|
||
_STAT_INTENT = re.compile(r"技术统计|统计|数据|谁得分|得分最高|表现|box|scorer|G\d|第.场")
|
||
|
||
|
||
def _attach_top_scorers(game_rows, max_games=3):
|
||
"""为比赛附加双方得分前三球员（供技术统计类问题）"""
|
||
for g in game_rows[:max_games]:
|
||
gid = g["id"]
|
||
rows = query("""SELECT gs.points, p.name AS player_name, t.name AS team_name
|
||
FROM game_player_stats gs
|
||
JOIN players p ON gs.player_id=p.id JOIN teams t ON gs.team_id=t.id
|
||
WHERE gs.game_id=? ORDER BY gs.points DESC LIMIT 6""", (gid,))
|
||
if rows:
|
||
g["top_scorers"] = [{"player": r["player_name"], "team": r["team_name"], "points": r["points"]}
|
||
for r in rows]
|
||
|
||
|
||
# ================================================================== 球队
|
||
def search_teams(query_text, limit=MAX_SHOW):
|
||
q = expand_aliases((query_text or "").strip())
|
||
if not q:
|
||
rows = query("SELECT * FROM teams ORDER BY name LIMIT ?", (limit,))
|
||
else:
|
||
seen, out = {}, []
|
||
sql = """SELECT * FROM teams WHERE name LIKE ? ESCAPE '\\' OR name_en LIKE ? ESCAPE '\\'
|
||
OR code LIKE ? ESCAPE '\\' OR city LIKE ? ESCAPE '\\' ORDER BY name LIMIT ?"""
|
||
for term in _split_terms(q)[:4]:
|
||
rows = _match_by_truncation(sql, term, lambda like: (like, like, like, like, limit))
|
||
for r in rows:
|
||
if r["id"] not in seen:
|
||
seen[r["id"]] = r
|
||
out.append(r)
|
||
rows = out[:limit]
|
||
return {"_source": "teams", "results": [_fmt_team(r) for r in rows]}
|
||
|
||
|
||
def get_team(team_id):
|
||
t = query_one("SELECT * FROM teams WHERE id=?", (team_id,))
|
||
if not t:
|
||
return None
|
||
return _fmt_team(t)
|
||
|
||
|
||
# ================================================================== 球员
|
||
def search_players(query_text, limit=MAX_SHOW):
|
||
q = expand_aliases((query_text or "").strip())
|
||
if not q:
|
||
rows = query("""SELECT p.*, t.name AS team_name FROM players p LEFT JOIN teams t ON p.team_id=t.id
|
||
ORDER BY p.season_pts DESC LIMIT ?""", (limit,))
|
||
return {"_source": "players", "results": [_fmt_player(r) for r in rows]}
|
||
seen, out = {}, []
|
||
sql = """SELECT p.*, t.name AS team_name FROM players p
|
||
LEFT JOIN teams t ON p.team_id=t.id
|
||
WHERE p.name LIKE ? ESCAPE '\\' OR p.name_en LIKE ? ESCAPE '\\'
|
||
OR t.name LIKE ? ESCAPE '\\'
|
||
ORDER BY p.season_pts DESC LIMIT ?"""
|
||
for term in _split_terms(q)[:4]:
|
||
rows = _match_by_truncation(sql, term, lambda like: (like, like, like, limit))
|
||
for r in rows:
|
||
if r["id"] not in seen:
|
||
seen[r["id"]] = r
|
||
out.append(r)
|
||
return {"_source": "players", "results": [_fmt_player(r) for r in out[:limit]]}
|
||
|
||
|
||
def get_player(player_id):
|
||
p = query_one("""SELECT p.*, t.name AS team_name FROM players p LEFT JOIN teams t ON p.team_id=t.id
|
||
WHERE p.id=?""", (player_id,))
|
||
return _fmt_player(p) if p else None
|
||
|
||
|
||
# ================================================================== 球员列表（前台分页版）
|
||
def list_players(q="", position="", page=1, size=24):
|
||
"""前台球员列表：关键词 + 位置筛选 + 分页（区别于 search_players 工具版）"""
|
||
q = expand_aliases((q or "").strip())
|
||
conds, args = [], []
|
||
if q:
|
||
like = f"%{fuzzy(q)}%"
|
||
conds.append("(p.name LIKE ? ESCAPE '\\' OR p.name_en LIKE ? ESCAPE '\\' OR t.name LIKE ? ESCAPE '\\')")
|
||
args += [like, like, like]
|
||
if position:
|
||
conds.append("p.position=?")
|
||
args.append(position)
|
||
where = ("WHERE " + " AND ".join(conds)) if conds else ""
|
||
total = query_one(f"SELECT COUNT(*) AS c FROM players p LEFT JOIN teams t ON p.team_id=t.id {where}", args)["c"]
|
||
rows = query(f"""SELECT p.*, t.name AS team_name FROM players p LEFT JOIN teams t ON p.team_id=t.id
|
||
{where} ORDER BY p.season_pts DESC LIMIT ? OFFSET ?""",
|
||
args + [size, (page - 1) * size])
|
||
return {"results": [_fmt_player(r) for r in rows], "total": total, "page": page, "size": size}
|
||
|
||
|
||
# ================================================================== 比赛
|
||
_TEAM_PAT = None
|
||
|
||
|
||
def _match_team_ids(text):
|
||
"""从文本中找出命中的球队名（支持中文/英文/缩写）"""
|
||
global _TEAM_PAT
|
||
teams = query("SELECT id, name, name_en, code FROM teams")
|
||
hits = []
|
||
for t in teams:
|
||
names = [t["name"], t["name_en"], t["code"]]
|
||
for n in names:
|
||
if n and len(n) >= 2 and n.lower() in (text or "").lower():
|
||
hits.append(t["id"])
|
||
break
|
||
return hits
|
||
|
||
|
||
def search_games(query_text, limit=10):
|
||
q = (query_text or "").strip()
|
||
now = datetime.now()
|
||
sql = """SELECT g.*, ht.name AS home_name, ht.code AS home_code,
|
||
at.name AS away_name, at.code AS away_code
|
||
FROM games g JOIN teams ht ON g.home_team_id=ht.id JOIN teams at ON g.away_team_id=at.id"""
|
||
conds, args = [], []
|
||
|
||
tids = _match_team_ids(q)
|
||
if tids:
|
||
marks = ",".join("?" for _ in tids)
|
||
conds.append(f"(g.home_team_id IN ({marks}) OR g.away_team_id IN ({marks}))")
|
||
args += tids * 2
|
||
if q and ("已结束" in q or "结束" in q or "比分" in q or "结果" in q or "谁赢" in q):
|
||
conds.append("g.status='finished'")
|
||
if q and ("未开始" in q or "即将" in q or "赛程" in q or "预告" in q):
|
||
conds.append("g.status='scheduled'")
|
||
if q and "总决赛" in q:
|
||
conds.append("g.round_name='总决赛'")
|
||
if q and "季后赛" in q:
|
||
conds.append("g.round_name LIKE '季后赛%' OR g.round_name IN ('总决赛','东部决赛','西部决赛')")
|
||
|
||
# 时间语义
|
||
order = "g.game_time DESC"
|
||
if "总决赛" in q or "系列赛" in q:
|
||
order = "g.game_time ASC" # 系列赛按时间正序，便于模型按场次引用
|
||
if re.search(r"最近|最新|上一场|上一轮", q):
|
||
conds.append("g.status='finished'")
|
||
elif re.search(r"下一场|即将|赛程|预告|未来", q):
|
||
conds.append("g.status='scheduled'")
|
||
order = "g.game_time ASC"
|
||
elif re.search(r"明天|明日", q):
|
||
conds.append("date(g.game_time)=date('now','localtime','+1 day')")
|
||
elif re.search(r"今天|今日", q):
|
||
conds.append("date(g.game_time)=date('now','localtime')")
|
||
|
||
where = ("WHERE " + " AND ".join(conds)) if conds else ""
|
||
sql += f" {where} ORDER BY {order} LIMIT ?"
|
||
args.append(limit)
|
||
rows = query(sql, args)
|
||
results = [_fmt_game(r) for r in rows]
|
||
if _STAT_INTENT.search(q):
|
||
# 若点名了具体场次（G6/第六场/日期），只保留该场并附完整得分榜
|
||
specific = re.search(r"G(\d)|第([一二三四五六七八九十])场|(\d{1,2})月(\d{1,2})日|(\d{4}-\d{2}-\d{2})", q)
|
||
if specific:
|
||
key = (specific.group(1) or "")
|
||
kept = []
|
||
for g in results:
|
||
gno = g["game_time"][8:10]
|
||
if key:
|
||
try:
|
||
gno = str(int(g["game_time"][8:10]))
|
||
except Exception:
|
||
gno = ""
|
||
if gno == key:
|
||
kept.append(g)
|
||
elif specific.group(5):
|
||
if g["game_time"][:10] == specific.group(5):
|
||
kept.append(g)
|
||
elif specific.group(3):
|
||
if g["game_time"][5:7] == specific.group(3) and g["game_time"][8:10] == specific.group(4):
|
||
kept.append(g)
|
||
if kept:
|
||
results = kept
|
||
_attach_top_scorers(results, max_games=len(results))
|
||
return {"_source": "games", "results": results}
|
||
|
||
|
||
def get_game_detail(game_id):
|
||
g = query_one("""SELECT g.*, ht.name AS home_name, ht.code AS home_code,
|
||
at.name AS away_name, at.code AS away_code
|
||
FROM games g JOIN teams ht ON g.home_team_id=ht.id JOIN teams at ON g.away_team_id=at.id
|
||
WHERE g.id=?""", (game_id,))
|
||
if not g:
|
||
return None
|
||
detail = _fmt_game(g)
|
||
stats = query("""SELECT gs.*, p.name AS player_name, p.position, t.name AS team_name
|
||
FROM game_player_stats gs
|
||
JOIN players p ON gs.player_id=p.id JOIN teams t ON gs.team_id=t.id
|
||
WHERE gs.game_id=? ORDER BY gs.team_id, gs.points DESC""", (game_id,))
|
||
detail["box_score"] = stats
|
||
return detail
|
||
|
||
|
||
# ================================================================== 比赛列表（前台分页版）
|
||
def list_games(q="", status="", page=1, size=12):
|
||
"""前台比赛列表：关键词 + 状态筛选 + 分页（数据量小，全量匹配后切片）"""
|
||
r = search_games(q, limit=500)
|
||
results = r.get("results", [])
|
||
if status:
|
||
results = [g for g in results if g["status"] == status]
|
||
total = len(results)
|
||
start = (page - 1) * size
|
||
return {"results": results[start:start + size], "total": total, "page": page, "size": size}
|
||
|
||
|
||
# ================================================================== 排名
|
||
def search_standings(query_text, limit=20, season=None):
|
||
q = (query_text or "").strip()
|
||
tids = _match_team_ids(q)
|
||
sql = """SELECT s.*, t.name AS team_name, t.code
|
||
FROM standings s JOIN teams t ON s.team_id=t.id"""
|
||
conds, args = [], []
|
||
if season:
|
||
conds.append("s.season=?")
|
||
args.append(season)
|
||
if tids:
|
||
marks = ",".join("?" for _ in tids)
|
||
conds.append(f"s.team_id IN ({marks})")
|
||
args += tids
|
||
if "东部" in q:
|
||
conds.append("s.conference='东部'")
|
||
if "西部" in q:
|
||
conds.append("s.conference='西部'")
|
||
where = ("WHERE " + " AND ".join(conds)) if conds else ""
|
||
rows = query(f"{sql} {where} ORDER BY s.conference DESC, s.rank ASC LIMIT ?", args + [limit])
|
||
return {"_source": "standings", "results": [
|
||
{"team": r["team_name"], "team_id": r["team_id"], "code": r["code"], "conference": r["conference"], "rank": r["rank"],
|
||
"wins": r["wins"], "losses": r["losses"], "win_pct": round(r["win_pct"] * 100, 1),
|
||
"season": r["season"]} for r in rows]}
|
||
|
||
|
||
# ================================================================== 比赛结果（按赛季）
|
||
def team_games(team_id, season=None, limit=15):
|
||
"""球队参与的比赛（详情页用），可按赛季过滤"""
|
||
conds, args = ["(g.home_team_id=? OR g.away_team_id=?)"], [team_id, team_id]
|
||
if season:
|
||
conds.append("g.season=?")
|
||
args.append(season)
|
||
rows = query(f"""SELECT g.*, ht.name AS home_name, ht.code AS home_code,
|
||
at.name AS away_name, at.code AS away_code
|
||
FROM games g JOIN teams ht ON g.home_team_id=ht.id JOIN teams at ON g.away_team_id=at.id
|
||
WHERE {" AND ".join(conds)} ORDER BY g.game_time DESC LIMIT ?""",
|
||
args + [limit])
|
||
return [_fmt_game(r) for r in rows]
|
||
|
||
|
||
# ================================================================== 球员比赛记录（技术统计）
|
||
def player_game_logs(player_id, limit=10):
|
||
"""球员参与的比赛与单场数据（详情页用）"""
|
||
rows = query("""SELECT gs.*, g.round_name, g.game_time, g.home_score, g.away_score,
|
||
ht.name AS home_name, at.name AS away_name
|
||
FROM game_player_stats gs
|
||
JOIN games g ON gs.game_id=g.id
|
||
JOIN teams ht ON g.home_team_id=ht.id JOIN teams at ON g.away_team_id=at.id
|
||
WHERE gs.player_id=? ORDER BY g.game_time DESC LIMIT ?""", (player_id, limit))
|
||
out = []
|
||
for r in rows:
|
||
out.append({"game_id": r["game_id"], "round_name": r["round_name"], "game_time": r["game_time"],
|
||
"home_team": r["home_name"], "away_team": r["away_name"],
|
||
"home_score": r["home_score"], "away_score": r["away_score"],
|
||
"pts": r["points"], "reb": r["rebounds"], "ast": r["assists"],
|
||
"stl": r["steals"], "blk": r["blocks"], "min": r["minutes"]})
|
||
return out
|
||
|
||
|
||
# ================================================================== 新闻列表（前台分页版）
|
||
def list_news(q="", kind="", tag="", page=1, size=10):
|
||
"""前台新闻列表：关键词 + 类型(kind) + 话题标签(tag) + 分页"""
|
||
conds, args = [], []
|
||
if q:
|
||
like = f"%{fuzzy(q)}%"
|
||
conds.append("(title LIKE ? ESCAPE '\\' OR content LIKE ? ESCAPE '\\' OR tags LIKE ? ESCAPE '\\')")
|
||
args += [like, like, like]
|
||
if kind:
|
||
conds.append("kind=?")
|
||
args.append(kind)
|
||
if tag:
|
||
conds.append("tags LIKE ? ESCAPE '\\'")
|
||
args.append(f"%{fuzzy(tag)}%")
|
||
where = ("WHERE " + " AND ".join(conds)) if conds else ""
|
||
total = query_one(f"SELECT COUNT(*) AS c FROM news {where}", args)["c"]
|
||
rows = query(f"""SELECT id,title,author,source,publish_time,tags,kind,substr(content,1,160) AS summary
|
||
FROM news {where} ORDER BY publish_time DESC, id DESC LIMIT ? OFFSET ?""",
|
||
args + [size, (page - 1) * size])
|
||
return {"results": rows, "total": total, "page": page, "size": size}
|
||
|
||
|
||
# ================================================================== 新闻（SQL 关键词 + 向量语义 + 可选 rerank 融合）
|
||
def search_news(query_text, limit=5):
|
||
q = (query_text or "").strip()
|
||
seen, out = set(), []
|
||
|
||
|
||
# ================================================================== 新闻（SQL 关键词 + 向量语义 + 可选 rerank 融合）
|
||
def search_news(query_text, limit=5):
|
||
q = (query_text or "").strip()
|
||
seen, out = set(), []
|
||
# 1) SQL 关键词命中（标题+正文+标签）
|
||
if q:
|
||
like = f"%{fuzzy(q)}%"
|
||
rows = query("""SELECT * FROM news WHERE kind='news' AND (title LIKE ? ESCAPE '\\' OR content LIKE ? ESCAPE '\\'
|
||
OR tags LIKE ? ESCAPE '\\') ORDER BY publish_time DESC LIMIT ?""",
|
||
(like, like, like, limit))
|
||
for r in rows:
|
||
seen.add(r["id"])
|
||
out.append({"id": r["id"], "title": r["title"], "content": r["content"][:220],
|
||
"publish_time": r["publish_time"], "source": r["source"], "tags": r["tags"]})
|
||
# 2) 向量语义补充
|
||
try:
|
||
hits = vector_store.query_vectors(q, n_results=limit * 3)
|
||
cands = []
|
||
for h in hits:
|
||
nid = h["metadata"].get("news_id")
|
||
if nid in seen or h["metadata"].get("kind") != "news":
|
||
continue
|
||
n = query_one("SELECT * FROM news WHERE id=?", (nid,))
|
||
if n:
|
||
cands.append({"id": nid, "title": n["title"], "content": n["content"][:220],
|
||
"publish_time": n["publish_time"], "source": n["source"],
|
||
"tags": n["tags"], "_score": h["distance"]})
|
||
cands.sort(key=lambda x: x["_score"])
|
||
for c in cands[: max(0, limit - len(out))]:
|
||
out.append({k: v for k, v in c.items() if k != "_score"})
|
||
except Exception as e:
|
||
pass
|
||
return {"_source": "news", "results": out}
|
||
|
||
|
||
# ================================================================== 人物
|
||
def search_persons(query_text, limit=MAX_SHOW):
|
||
q = (query_text or "").strip()
|
||
if not q:
|
||
rows = query("""SELECT p.*, t.name AS team_name FROM persons p LEFT JOIN teams t ON p.team_id=t.id
|
||
ORDER BY p.id LIMIT ?""", (limit,))
|
||
else:
|
||
like = f"%{fuzzy(q)}%"
|
||
rows = query("""SELECT p.*, t.name AS team_name FROM persons p LEFT JOIN teams t ON p.team_id=t.id
|
||
WHERE p.name LIKE ? ESCAPE '\\' OR p.name_en LIKE ? ESCAPE '\\'
|
||
OR p.role LIKE ? ESCAPE '\\' OR p.role_cn LIKE ? ESCAPE '\\'
|
||
OR t.name LIKE ? ESCAPE '\\' OR p.bio LIKE ? ESCAPE '\\'
|
||
ORDER BY p.id LIMIT ?""", (like, like, like, like, like, like, limit))
|
||
return {"_source": "persons", "results": [
|
||
{"id": r["id"], "name": r["name"], "name_en": r["name_en"], "role": r["role"],
|
||
"role_cn": r["role_cn"], "title": r["title"], "team": r["team_name"],
|
||
"bio": r["bio"], "achievements": r["achievements"]} for r in rows]}
|
||
|
||
|
||
# ================================================================== 知识百科（纯向量检索）
|
||
def search_knowledge(query_text, limit=3):
|
||
q = (query_text or "").strip()
|
||
try:
|
||
hits = vector_store.query_vectors(q, n_results=limit)
|
||
except Exception:
|
||
return {"_source": "knowledge", "results": []}
|
||
out = []
|
||
for h in hits:
|
||
if h["metadata"].get("kind") != "wiki":
|
||
continue
|
||
n = query_one("SELECT * FROM news WHERE id=?", (h["metadata"].get("news_id"),))
|
||
if n:
|
||
out.append({"title": n["title"], "content": n["content"][:400]})
|
||
return {"_source": "knowledge", "results": out}
|
||
|
||
|
||
# ================================================================== 工具注册表（供 LLM function calling）
|
||
TOOLS = [
|
||
{"type": "function", "function": {"name": "search_teams", "description": "查询球队信息（名称/城市/主场/主教练/总冠军数），支持中文名、英文名、缩写",
|
||
"parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "球队名或关键词，如'湖人'或'Lakers'"}, "limit": {"type": "integer", "description": "返回条数，默认6"}}, "required": ["query"]}}},
|
||
{"type": "function", "function": {"name": "search_players", "description": "查询球员信息（球队/位置/本赛季与生涯数据/荣誉），支持中英文名或球队名",
|
||
"parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "球员名或球队名，如'库里'或'Stephen Curry'"}, "limit": {"type": "integer", "description": "返回条数，默认6"}}, "required": ["query"]}}},
|
||
{"type": "function", "function": {"name": "search_games", "description": "查询比赛信息（比分/时间/轮次），支持球队名+时间语义（最近/上一场/下一场/今天/明天），如'湖人最近比赛'、'总决赛第六场'",
|
||
"parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "球队名或时间描述，如'勇士 最近'"}, "limit": {"type": "integer", "description": "返回条数，默认10"}}, "required": ["query"]}}},
|
||
{"type": "function", "function": {"name": "get_game_detail", "description": "获取单场比赛详情及双方球员技术统计（得分/篮板/助攻等），参数为比赛ID",
|
||
"parameters": {"type": "object", "properties": {"game_id": {"type": "integer", "description": "比赛ID（先调用search_games获得）"}}, "required": ["game_id"]}}},
|
||
{"type": "function", "function": {"name": "search_standings", "description": "查询球队排名（东西部/胜场/胜率），如'西部排名'或'湖人战绩'",
|
||
"parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "球队名或'西部'/'东部'"}}, "required": ["query"]}}},
|
||
{"type": "function", "function": {"name": "search_news", "description": "查询新闻资讯（交易/伤病/奖项/动态），支持关键词或语义描述",
|
||
"parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "新闻关键词或话题，如'选秀'、'詹姆斯续约'"}, "limit": {"type": "integer", "description": "返回条数，默认5"}}, "required": ["query"]}}},
|
||
{"type": "function", "function": {"name": "search_persons", "description": "查询篮球相关人物（教练/经纪人/评论员/主持人/总经理/传奇），如'波波维奇'、'杨毅'",
|
||
"parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "人名或角色，如'勇士主教练'、'评论员'"}, "limit": {"type": "integer", "description": "返回条数，默认6"}}, "required": ["query"]}}},
|
||
{"type": "function", "function": {"name": "search_knowledge", "description": "查询NBA知识百科（历史/规则/制度/纪录等），如'工资帽'、'选秀制度'、'三分球历史'",
|
||
"parameters": {"type": "object", "properties": {"query": {"type": "string", "description": "知识话题描述"}, "limit": {"type": "integer", "description": "返回条数，默认3"}}, "required": ["query"]}}},
|
||
]
|
||
|
||
TOOL_HANDLERS = {
|
||
"search_teams": search_teams,
|
||
"search_players": search_players,
|
||
"search_games": search_games,
|
||
"get_game_detail": get_game_detail,
|
||
"search_standings": search_standings,
|
||
"search_news": search_news,
|
||
"search_persons": search_persons,
|
||
"search_knowledge": search_knowledge,
|
||
}
|
||
|
||
|
||
def run_tool(name, args_dict):
|
||
"""执行工具调用（统一异常兜底，schema参数名→函数参数名映射）"""
|
||
try:
|
||
fn = TOOL_HANDLERS.get(name)
|
||
if not fn:
|
||
return {"_source": "error", "results": [], "error": f"未知工具 {name}"}
|
||
args = dict(args_dict or {})
|
||
if "query" in args and "query_text" in fn.__code__.co_varnames:
|
||
args["query_text"] = args.pop("query")
|
||
return fn(**args)
|
||
except Exception as e:
|
||
return {"_source": "error", "results": [], "error": f"工具执行失败: {e}"}
|