Spaces:
Running
Running
File size: 5,147 Bytes
275d495 36ef4c6 275d495 36ef4c6 275d495 36ef4c6 275d495 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """Deterministic guardrails for deciding whether another Skill is necessary."""
from __future__ import annotations
_EXPLICIT_SKILL_MARKERS: dict[str, tuple[str, ...]] = {
"announcement-search": (
"公告", "披露", "财报", "季报", "年报", "业绩预告",
"announcement", "filing",
),
"news-search": (
"新闻", "资讯", "快讯", "舆情", "动态", "近况", "消息面", "news",
),
"report-search": (
"研报", "研究报告", "券商观点", "机构观点", "research report",
),
}
_CAUSAL_OR_EVENT_ANALYSIS_MARKERS = (
"为什么", "原因", "怎么回事", "影响", "利好", "利空", "风险",
"消息面", "催化", "异动",
)
_DECISION_ANALYSIS_MARKERS = (
"投资价值", "值得买", "能买吗", "能不能买", "是否买", "该不该买",
"下周一买", "买入", "卖出", "推荐哪些", "推荐哪",
)
_BROAD_ANALYSIS_MARKERS = (
"综合分析", "全面分析", "深度分析", "多维分析", "各方面", "等等",
)
_VERTICAL_SKILL_ORDER = (
"news-search",
"announcement-search",
"report-search",
)
_DIRECT_FINANCIAL_DATA_MARKERS = (
"股票", "股价", "行情", "市值", "涨停", "跌停", "涨幅", "跌幅",
"成交", "换手", "资金流", "走势", "技术面", "基金", "指数", "可转债", "期货",
"营收", "营业收入", "净利润", "财务指标", "交易日", "开盘",
"市盈率", "市净率", "股息率", "分红率", "毛利率", "净利率",
"每股收益", "净资产收益率", "资产负债率", "现金流",
)
def has_cross_source_analysis_intent(user_query: str) -> bool:
"""Whether the user asked for analysis that can justify source expansion.
Bare movement words such as ``涨`` and ``跌`` are intentionally excluded:
they also occur in simple data requests like ``列举全部涨停股票``.
"""
return bool(required_vertical_skills_for_query(user_query))
def explicitly_requests_skill(user_query: str, skill: str) -> bool:
"""Return True only when the latest question names that evidence family."""
q = (user_query or "").lower()
return any(marker.lower() in q for marker in _EXPLICIT_SKILL_MARKERS.get(skill, ()))
def required_vertical_skills_for_query(user_query: str) -> tuple[str, ...]:
"""Map user-requested evidence dimensions to the owning search Skills.
The mapping follows each registered Skill's responsibility:
news supplies public information, announcements supply first-party company
events, and reports supply analyst/ratings evidence. A broad company
analysis needs all three; causal analysis needs news plus announcements.
"""
q = (user_query or "").lower()
required = {
skill for skill in _VERTICAL_SKILL_ORDER if explicitly_requests_skill(q, skill)
}
if (
"news-search" not in required
and {"announcement-search", "report-search"}.issubset(required)
and any(marker in q for marker in ("或", "或者", "二选一", "任一"))
):
# One vertical source is sufficient when the user explicitly offers
# announcements or reports as alternatives. Keep announcement-first
# behavior for consistency with the existing product contract.
required.discard("report-search")
causal = any(marker.lower() in q for marker in _CAUSAL_OR_EVENT_ANALYSIS_MARKERS)
decision = any(marker.lower() in q for marker in _DECISION_ANALYSIS_MARKERS)
broad = any(marker.lower() in q for marker in _BROAD_ANALYSIS_MARKERS) or (
"对" in q and "的分析" in q
)
if causal:
required.update(("news-search", "announcement-search"))
if decision or broad:
required.update(_VERTICAL_SKILL_ORDER)
return tuple(skill for skill in _VERTICAL_SKILL_ORDER if skill in required)
def is_skill_necessary_for_query(user_query: str, skill: str) -> bool:
"""Validate that a planned/reflected Skill serves the latest question.
Structured financial queries remain generally valid. Public web search is
retained as an empty-result fallback. Vertical financial searches require
either an explicit request for that source or a genuine analysis/decision
intent; merely discovering a stock name never authorizes extra lookups.
"""
if skill in {"financial-query", "anysearch"}:
return True
if skill in _EXPLICIT_SKILL_MARKERS:
return skill in required_vertical_skills_for_query(user_query)
return True
def requests_financial_data(user_query: str) -> bool:
"""Recognize financial metrics independently of news/analysis dimensions."""
q = (user_query or "").lower()
return any(marker.lower() in q for marker in _DIRECT_FINANCIAL_DATA_MARKERS)
def is_direct_financial_data_query(user_query: str) -> bool:
"""Whether one or more structured financial lookups can answer the query."""
if not requests_financial_data(user_query):
return False
if has_cross_source_analysis_intent(user_query):
return False
return not any(explicitly_requests_skill(user_query, skill) for skill in _EXPLICIT_SKILL_MARKERS)
|