388 lines
16 KiB
Python
388 lines
16 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from functools import lru_cache
|
||
from typing import Any
|
||
|
||
from backend.bootstrap.config import APP_DIR
|
||
|
||
|
||
KNOWLEDGE_FILE = APP_DIR / "data" / "heaven_knowledge.json"
|
||
|
||
|
||
def prepare_heaven_context(mode: str, calculation: dict[str, Any]) -> dict[str, Any]:
|
||
"""Build the only context shape that may cross the LLM boundary."""
|
||
if mode == "trend":
|
||
prepared = _prepare_trend(calculation)
|
||
elif mode == "fortune":
|
||
prepared = _prepare_fortune(calculation)
|
||
elif mode == "heart":
|
||
prepared = _prepare_heart(calculation)
|
||
else:
|
||
raise ValueError("不支持的问天知识模式。")
|
||
prepared["knowledge"] = retrieve_heaven_knowledge(mode, prepared)
|
||
return prepared
|
||
|
||
|
||
def retrieve_heaven_knowledge(mode: str, context: dict[str, Any]) -> dict[str, Any]:
|
||
catalog = _knowledge_catalog()
|
||
source_ids: list[str]
|
||
records: list[dict[str, Any]]
|
||
if mode == "trend":
|
||
source_ids = ["zhouyi"]
|
||
records = _trend_records(catalog, context)
|
||
elif mode == "fortune":
|
||
source_ids = ["neijing"]
|
||
records = _fortune_records(catalog, context)
|
||
elif mode == "heart":
|
||
source_ids = ["zhouyi", "jingfang", "huozhulin", "zengshan"]
|
||
records = _heart_records(catalog, context)
|
||
else:
|
||
raise ValueError("不支持的问天知识模式。")
|
||
return {
|
||
"version": str(catalog.get("version") or ""),
|
||
"retrieval": "deterministic-keyed",
|
||
"sources": [
|
||
{"id": source_id, **dict(catalog["sources"][source_id])}
|
||
for source_id in source_ids
|
||
],
|
||
"records": records,
|
||
}
|
||
|
||
|
||
def _prepare_trend(context: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"mode": "trend",
|
||
"calculation": {
|
||
"data_trade_date": context.get("data_trade_date") or "",
|
||
"hexagram": context.get("hexagram") or {},
|
||
"movement": context.get("movement") or {},
|
||
},
|
||
"interpretation_contract": {
|
||
"required": ["明确卦势倾向", "主要矛盾", "实际动爻转折", "本卦到之卦的变化关系"],
|
||
"allowed": ["偏进或偏守", "先难后易或由盛转收", "结论有条件或存在分歧"],
|
||
"forbidden": ["原始行情旁证", "具体涨跌预测", "时间点预测", "无条件买卖指令", "泛化劝诫代替解卦"],
|
||
},
|
||
}
|
||
|
||
|
||
def _prepare_fortune(context: dict[str, Any]) -> dict[str, Any]:
|
||
field = context.get("five_phase_field") or {}
|
||
framework = field.get("framework") or {}
|
||
relations = framework.get("relations") or {}
|
||
layers = {
|
||
str(item.get("id") or ""): item
|
||
for item in framework.get("layers") or []
|
||
if isinstance(item, dict)
|
||
}
|
||
six_qi = field.get("six_qi") or {}
|
||
movement = field.get("movement") or {}
|
||
pillars = field.get("pillars") or {}
|
||
personal = context.get("personal_profile") or {}
|
||
sector_catalog = {
|
||
str(group.get("element") or ""): [
|
||
str(item.get("name") or "").strip()
|
||
for item in group.get("industries") or []
|
||
if str(item.get("name") or "").strip()
|
||
]
|
||
for group in field.get("sector_catalog") or []
|
||
if isinstance(group, dict)
|
||
}
|
||
industry_bases: dict[str, list[str]] = {}
|
||
|
||
def add_industry_basis(element: str, basis: str) -> None:
|
||
if element not in sector_catalog or not sector_catalog[element]:
|
||
return
|
||
industry_bases.setdefault(element, [])
|
||
if basis not in industry_bases[element]:
|
||
industry_bases[element].append(basis)
|
||
|
||
add_industry_basis(str(movement.get("phase") or ""), "中运")
|
||
for label, qi in (
|
||
("司天", six_qi.get("sitian")),
|
||
("在泉", six_qi.get("zaiquan")),
|
||
("主气", six_qi.get("host_qi")),
|
||
("客气", six_qi.get("guest_qi")),
|
||
):
|
||
qi_text = str(qi or "")
|
||
add_industry_basis(qi_text[-1:] if qi_text else "", label)
|
||
day_master = personal.get("day_master") or {}
|
||
current = personal.get("current") or {}
|
||
personal_context = {}
|
||
if day_master:
|
||
current_ten_gods = current.get("ten_gods") or {}
|
||
personal_context = {
|
||
"natal_day_master": {
|
||
"stem": day_master.get("stem") or "",
|
||
"element": day_master.get("element") or "",
|
||
},
|
||
"today_relative_to_natal_day_master": {
|
||
"pillars": current.get("pillars") or {},
|
||
"stem_relations": {
|
||
key: str((current_ten_gods.get(key) or {}).get("stem") or "")
|
||
for key in ("year", "month", "day")
|
||
},
|
||
},
|
||
}
|
||
return {
|
||
"mode": "fortune",
|
||
"calculation": {
|
||
"calendar_date": context.get("calendar_date") or field.get("date") or "",
|
||
"lunar_date": field.get("lunar_date") or "",
|
||
"pillars": {
|
||
"year": pillars.get("year") or "",
|
||
"month": pillars.get("month") or "",
|
||
"day": pillars.get("day") or "",
|
||
},
|
||
"solar_terms": field.get("solar_terms") or {},
|
||
"year_movement": {
|
||
"phase": movement.get("phase") or "",
|
||
"tendency": movement.get("tendency") or "",
|
||
"label": movement.get("label") or "",
|
||
},
|
||
"annual_qi": {
|
||
"sitian": six_qi.get("sitian") or "",
|
||
"zaiquan": six_qi.get("zaiquan") or "",
|
||
"ruling": six_qi.get("ruling") or "",
|
||
"ruling_qi": six_qi.get("ruling_qi") or "",
|
||
"annual_pattern": relations.get("annual_pattern") or {},
|
||
},
|
||
"current_qi": {
|
||
"step": six_qi.get("step"),
|
||
"step_name": six_qi.get("step_name") or "",
|
||
"host_qi": six_qi.get("host_qi") or "",
|
||
"guest_qi": six_qi.get("guest_qi") or "",
|
||
"guest_host_relation": relations.get("guest_host") or {},
|
||
"alignment": relations.get("alignment") or six_qi.get("alignment") or "",
|
||
},
|
||
"day_trigger": {
|
||
"day_pillar": pillars.get("day") or "",
|
||
"summary": (layers.get("day") or {}).get("summary") or "",
|
||
},
|
||
"industry_symbols": [
|
||
{
|
||
"element": element,
|
||
"basis": bases,
|
||
"industries": sector_catalog[element],
|
||
}
|
||
for element, bases in industry_bases.items()
|
||
],
|
||
"personal": personal_context,
|
||
},
|
||
"excluded_from_interpretation": [
|
||
"五行权重与百分比",
|
||
"主导元素排序",
|
||
"权重生成的复合断语",
|
||
"预制情绪与交易行为结论",
|
||
"行业实时行情旁证",
|
||
"简化喜用神与强弱结论",
|
||
],
|
||
"interpretation_contract": {
|
||
"required": ["年纲", "当前客主加临", "日辰触发", "行业影响", "个人合参（如有）", "制衡动作"],
|
||
"forbidden": [
|
||
"重新计算五行权重",
|
||
"把相生直接判吉",
|
||
"把相克直接判凶",
|
||
"用市场涨跌证明气场",
|
||
"把行业取象写成行业涨跌预测或投资推荐",
|
||
"把当日日柱误称为用户命局",
|
||
"推断未提供的命局强弱或喜用神",
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
def _prepare_heart(context: dict[str, Any]) -> dict[str, Any]:
|
||
preset = str(context.get("question_preset") or "custom")
|
||
if preset not in {"trade", "mind", "unthemed", "custom"}:
|
||
preset = "custom"
|
||
return {
|
||
"mode": "heart",
|
||
"calculation": {
|
||
"question": str(context.get("question") or "").strip(),
|
||
"question_preset": preset,
|
||
"question_scope": {
|
||
"trade": "股票交易中的参与条件、机会、阻碍与风险，不是融资或商业合作问题。",
|
||
"mind": "影响股票交易判断的情绪、执念或盲点。",
|
||
"unthemed": "不指定事项的一般观照。",
|
||
"custom": "只按用户实际写出的事项理解，不补写背景。",
|
||
}[preset],
|
||
"ritual": context.get("ritual") or {},
|
||
"hexagram": context.get("hexagram") or {},
|
||
"six_yao": context.get("six_yao") or {},
|
||
},
|
||
"interpretation_contract": {
|
||
"required": ["回应所问", "本卦处境", "世应与相关六亲", "关键动变", "之卦趋向", "可验证动作"],
|
||
"plain_language": "专业术语首次出现时立即用白话解释。",
|
||
"forbidden": [
|
||
"修改纳甲排盘",
|
||
"猜测未输入的问题",
|
||
"把股票交易改写成融资或合作问题",
|
||
"把六亲直接等同于现实人物或资金来源",
|
||
"单凭六神或空亡断吉凶",
|
||
"根据旬空填实或干支日期预测应期",
|
||
"具体股价和时间点预测",
|
||
"无条件买卖指令",
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
def _trend_records(catalog: dict[str, Any], context: dict[str, Any]) -> list[dict[str, Any]]:
|
||
hexagram = (context.get("calculation") or {}).get("hexagram") or {}
|
||
moving = [line for line in hexagram.get("lines") or [] if line.get("moving")]
|
||
method_key = "stable" if not moving else "single" if len(moving) == 1 else "multiple"
|
||
rules = catalog["trend"]["rules"]
|
||
records = [
|
||
{"id": "trend-method", "source": "product_method", "text": catalog["trend"]["method"]},
|
||
{"id": f"trend-moving-{method_key}", "source": "product_method", "text": rules[method_key]},
|
||
_hexagram_record("primary", hexagram),
|
||
]
|
||
records.extend(_line_record(line) for line in moving)
|
||
transformed = hexagram.get("transformed") or {}
|
||
if transformed:
|
||
records.append(_hexagram_record("transformed", transformed))
|
||
return records
|
||
|
||
|
||
def _fortune_records(catalog: dict[str, Any], context: dict[str, Any]) -> list[dict[str, Any]]:
|
||
calculation = context.get("calculation") or {}
|
||
movement = calculation.get("year_movement") or {}
|
||
annual_qi = calculation.get("annual_qi") or {}
|
||
current_qi = calculation.get("current_qi") or {}
|
||
knowledge = catalog["fortune"]
|
||
records = [
|
||
{"id": "fortune-principle", "source": "neijing", "text": knowledge["principle"]},
|
||
]
|
||
if calculation.get("industry_symbols"):
|
||
records.append(
|
||
{
|
||
"id": "fortune-industry-boundary",
|
||
"source": "product_method",
|
||
"text": knowledge["industry_boundary"],
|
||
}
|
||
)
|
||
personal = calculation.get("personal") or {}
|
||
if personal:
|
||
records.append(
|
||
{
|
||
"id": "fortune-personal-boundary",
|
||
"source": "product_method",
|
||
"text": knowledge["personal_boundary"],
|
||
}
|
||
)
|
||
today = personal.get("today_relative_to_natal_day_master") or {}
|
||
relation_semantics = knowledge.get("personal_relations") or {}
|
||
for relation in dict.fromkeys((today.get("stem_relations") or {}).values()):
|
||
if relation in relation_semantics:
|
||
records.append(
|
||
{
|
||
"id": f"fortune-personal-{relation}",
|
||
"source": "product_method",
|
||
"subject": relation,
|
||
"text": relation_semantics[relation],
|
||
}
|
||
)
|
||
tendency = str(movement.get("tendency") or "")
|
||
if tendency in knowledge["movement"]:
|
||
records.append({"id": f"movement-{tendency}", "source": "neijing", "text": knowledge["movement"][tendency]})
|
||
for key in ("sitian", "zaiquan"):
|
||
qi = str(annual_qi.get(key) or "")
|
||
if qi in knowledge["qi"]:
|
||
records.append({"id": f"annual-{key}", "source": "neijing", "subject": qi, "text": knowledge["qi"][qi]})
|
||
for key in ("host_qi", "guest_qi"):
|
||
qi = str(current_qi.get(key) or "")
|
||
if qi in knowledge["qi"]:
|
||
records.append({"id": f"current-{key}", "source": "neijing", "subject": qi, "text": knowledge["qi"][qi]})
|
||
relation = current_qi.get("guest_host_relation") or {}
|
||
relation_type = str(relation.get("type") or "")
|
||
if relation_type in knowledge["relations"]:
|
||
records.append({"id": f"relation-{relation_type}", "source": "neijing", "subject": relation.get("label") or "", "text": knowledge["relations"][relation_type]})
|
||
records.append({"id": "day-trigger", "source": "neijing", "text": knowledge["day_trigger"]})
|
||
return records
|
||
|
||
|
||
def _heart_records(catalog: dict[str, Any], context: dict[str, Any]) -> list[dict[str, Any]]:
|
||
calculation = context.get("calculation") or {}
|
||
hexagram = calculation.get("hexagram") or {}
|
||
six_yao = calculation.get("six_yao") or {}
|
||
preset = str(calculation.get("question_preset") or "custom")
|
||
heart = catalog["heart"]
|
||
records = [
|
||
{"id": "heart-focus", "source": "product_method", "text": heart["focus"].get(preset, heart["focus"]["custom"])},
|
||
{"id": "heart-evidence-order", "source": "product_method", "items": heart["evidence_order"]},
|
||
{"id": "heart-limits", "source": "product_method", "text": heart["limits"]},
|
||
{"id": "heart-self-response", "source": "jingfang", "text": heart["semantics"]["self_response"]},
|
||
{"id": "heart-calendar", "source": "zengshan", "text": heart["semantics"]["calendar"]},
|
||
{"id": "heart-movement", "source": "huozhulin", "text": heart["semantics"]["movement"]},
|
||
{"id": "heart-six-spirits", "source": "zengshan", "text": heart["semantics"]["six_spirits"]},
|
||
{"id": "heart-timing-boundary", "source": "product_method", "text": heart["semantics"]["timing_boundary"]},
|
||
_hexagram_record("primary", hexagram),
|
||
]
|
||
relatives = {
|
||
str(line.get("relative") or "")
|
||
for line in six_yao.get("lines") or []
|
||
if line.get("relative")
|
||
}
|
||
for relative in sorted(relatives):
|
||
text = (heart["semantics"].get("relatives") or {}).get(relative)
|
||
if text:
|
||
records.append(
|
||
{
|
||
"id": f"heart-relative-{relative}",
|
||
"source": "huozhulin",
|
||
"subject": relative,
|
||
"text": text,
|
||
}
|
||
)
|
||
records.extend(_line_record(line) for line in hexagram.get("lines") or [] if line.get("moving"))
|
||
transformed = hexagram.get("transformed") or {}
|
||
if transformed:
|
||
records.append(_hexagram_record("transformed", transformed))
|
||
palace = six_yao.get("palace") or {}
|
||
records.append(
|
||
{
|
||
"id": "heart-palace",
|
||
"source": "jingfang",
|
||
"text": (
|
||
f"本卦归{palace.get('name') or '--'}、{palace.get('stage') or '--'}，"
|
||
f"世在{palace.get('self_position') or '--'}爻，应在{palace.get('response_position') or '--'}爻。"
|
||
),
|
||
}
|
||
)
|
||
return records
|
||
|
||
|
||
def _hexagram_record(kind: str, hexagram: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"id": f"zhouyi-{kind}",
|
||
"source": "zhouyi",
|
||
"kind": kind,
|
||
"name": hexagram.get("name") or "",
|
||
"inner_trigram": hexagram.get("inner_trigram") or "",
|
||
"outer_trigram": hexagram.get("outer_trigram") or "",
|
||
"text": hexagram.get("text") or "",
|
||
"tuan": hexagram.get("tuan") or "",
|
||
"image": hexagram.get("image") or "",
|
||
}
|
||
|
||
|
||
def _line_record(line: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"id": f"zhouyi-line-{line.get('position') or ''}",
|
||
"source": "zhouyi",
|
||
"position": line.get("position"),
|
||
"position_name": line.get("position_name") or "",
|
||
"line_name": line.get("line_name") or "",
|
||
"text": line.get("text") or "",
|
||
"image": line.get("image") or "",
|
||
}
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def _knowledge_catalog() -> dict[str, Any]:
|
||
payload = json.loads(KNOWLEDGE_FILE.read_text(encoding="utf-8"))
|
||
if not payload.get("version") or not isinstance(payload.get("sources"), dict):
|
||
raise ValueError("问天知识库格式不完整。")
|
||
return payload
|