from __future__ import annotations import unittest from types import SimpleNamespace from unittest.mock import patch from backend.features.heaven.agent import ( HeavenAgentError, _system_prompt, _validate_answer, interpret_heaven, ) class HeavenPromptTests(unittest.TestCase): def test_trend_prompt_requires_direction_without_fixed_advice_vocabulary(self): prompt = _system_prompt("trend") self.assertIn("必须明确给出卦义上的当下倾向", prompt) self.assertIn("不得用“谨慎、等待、守信、辨伪”一类泛化劝诫代替解势", prompt) self.assertNotIn("卦理约占九成", prompt) self.assertNotIn("全文控制在300至450", prompt) self.assertIn("## 核心判断", prompt) self.assertIn("## 决策映射", prompt) def test_fortune_prompt_excludes_weighted_conclusion(self): prompt = _system_prompt("fortune") self.assertIn("五行权重、主导元素和预制复合断语已明确排除", prompt) self.assertIn("相生不直接等于吉，相克不直接等于凶", prompt) self.assertIn("不得使用藏干", prompt) self.assertIn("只能使用knowledge中本次命中的关系释义", prompt) self.assertNotIn("全文控制在420至600", prompt) self.assertIn("calculation.industry_symbols", prompt) self.assertIn("## 行业影响", prompt) self.assertIn("不得预测行业涨跌", prompt) def test_heart_prompt_uses_question_and_deterministic_chart(self): prompt = _system_prompt("heart") self.assertIn("用户在起卦前确定的问题位于calculation.question", prompt) self.assertIn("纳甲、卦宫、世应、六亲、六神", prompt) self.assertIn("不是商业合作、融资、借贷或寻找资金方", prompt) self.assertIn("不得假定用户已经入场", prompt) self.assertIn("不得制造外围资金、外围资源", prompt) self.assertIn("不得输出使用竖线分栏的Markdown表格", prompt) self.assertIn("不得用旬空、填实、出空或干支日推算", prompt) self.assertNotIn("最后三句极短的问心句", prompt) self.assertIn("## 所问之答", prompt) self.assertIn("## 可验证之处", prompt) def test_fortune_validator_requires_industry_section_when_symbols_are_present(self): context = { "calculation": { "industry_symbols": [ {"element": "水", "basis": ["中运"], "industries": ["航运"]} ] } } answer = "年纲与客主关系已经说明，日辰只作当日触发，最后给出个人观察和制衡动作。" * 2 with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer, context) _validate_answer("fortune", answer + "行业影响只作航运等水属性行业的传统取象，不预测涨跌。", context) def test_trend_validator_rejects_market_story_added_after_casting(self): answer = "本卦显示当下偏守，动爻揭示由增益转入受阻，之卦说明变化后应先稳住结构。" * 2 _validate_answer("trend", answer, {}) with self.assertRaises(HeavenAgentError): _validate_answer("trend", answer + "白酒行业政策可能带来外部压力。", {}) def test_fortune_validator_rejects_invented_natal_strength(self): context = { "calculation": { "pillars": {"year": "丙午", "month": "乙未", "day": "庚戌"}, "personal": { "today_relative_to_natal_day_master": { "pillars": {"year": "丙午", "month": "乙未", "day": "庚戌"} } }, } } answer = "丙午年纲、乙未客主和庚戌日辰形成相互制约，个人合参只说明当日关系，不据此判断命局强弱。" * 2 _validate_answer("fortune", answer, context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "个人本身偏财较重。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "戊辰年进一步增加土气。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "午中藏丁火，所以内在火气被引动。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "戌为火库，内藏火之余气。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "戌属土、含太阳寒水之气。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "木生火、火生土，因此本命日主受到泄耗。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "戌月进一步加重土气。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "日柱庚戌本身就是金。", context) with self.assertRaises(HeavenAgentError): _validate_answer("fortune", answer + "戌的寒水属性使气机进一步收敛。", context) def test_invalid_first_draft_is_rewritten_before_returning(self): invalid = "本卦解释完整，但白酒行业政策和基本面正在改变外部环境，因此当下应偏守。" * 2 valid = "本卦显示当下偏守，实际动爻揭示由增益转入受阻，之卦说明变化后要先确认结构能否重新承接。" * 2 with patch( "backend.features.heaven.agent.llm_transport.chat_completion", side_effect=[ SimpleNamespace(content=invalid, latency_ms=1), SimpleNamespace(content=valid, latency_ms=1), ], ) as completion: result = interpret_heaven( "trend", {}, "key", "https://example.invalid", "test-model" ) self.assertEqual(result["answer"], valid) self.assertEqual(completion.call_count, 2) repair_messages = completion.call_args_list[1].kwargs["messages"] self.assertIn("未通过本地一致性校验", repair_messages[-1]["content"]) def test_heart_validator_rejects_invented_financing_and_timing(self): context = { "calculation": { "question": "关于我心中的这笔交易，此刻最需要看清的机会、阻碍与风险是什么？" } } answer = "本卦说明条件仍有矛盾，世应提示内外预期并不一致，动爻要求先核对退出条件，再看之卦所示的收束方向。" * 2 _validate_answer("heart", answer, context) with self.assertRaises(HeavenAgentError): _validate_answer("heart", answer + "对方资金提供方会转为合作方。", context) with self.assertRaises(HeavenAgentError): _validate_answer("heart", answer + "未来三日等到旬空填实后再行动。", context) def test_heart_validator_rejects_unstated_position_assumptions(self): context = { "calculation": { "question": "关于我心中的这笔交易，此刻最需要看清的机会、阻碍与风险是什么？" } } answer = "本卦说明条件尚未完全衔接，世应提示内外预期有差异，动爻要求先核对参与条件，再看变化后的风险边界。" * 2 _validate_answer("heart", answer, context) for term in ("持仓", "仓位", "建仓", "入场", "持有", "买入", "卖出", "止损", "止盈"): with self.subTest(term=term), self.assertRaises(HeavenAgentError): _validate_answer("heart", answer + f"用户当前的{term}安排已经明确。", context) def test_heart_validator_allows_position_language_when_question_states_it(self): context = {"calculation": {"question": "我已经持仓，这笔交易的止损条件是否清楚？"}} answer = "本卦所示的内外条件仍有差异，世应关系提醒先核对持仓依据；动爻显示原判断正在变化，止损条件应以用户已经写下的客观边界为准。" * 2 _validate_answer("heart", answer, context) def test_heart_validator_rejects_invented_resource_paths(self): context = {"calculation": {"question": "这笔交易最需要防范什么？"}} answer = "本卦说明条件尚未完全衔接，世应提示内外预期有差异，动爻要求先核对参与条件，再看变化后的风险边界。" * 2 for scenario in ( "外围资金可能进入。", "外围资源进入路径尚不清楚。", "资金来源会在之后出现。", "资源提供的安排已经成形。", ): with self.subTest(scenario=scenario), self.assertRaises(HeavenAgentError): _validate_answer("heart", answer + scenario, context) def test_heart_validator_rejects_markdown_tables(self): context = {"calculation": {"question": "这笔交易最需要防范什么？"}} answer = "本卦说明条件尚未完全衔接，世应提示内外预期有差异，动爻要求先核对参与条件，再看变化后的风险边界。" * 2 tables = ( answer + "\n| 项目 | 解释 |\n| --- | --- |\n| 世爻 | 当前条件 |", answer + "\n项目 | 解释\n---|---\n世爻 | 当前条件", ) for table in tables: with self.subTest(table=table), self.assertRaises(HeavenAgentError): _validate_answer("heart", table, context) def test_heart_invalid_first_draft_is_rewritten_before_returning(self): context = {"calculation": {"question": "这笔交易最需要防范什么？"}} invalid = "本卦提示外围资源进入路径尚未落实，用户应先调整仓位，再等待资金进入。" * 3 valid = "本卦说明条件尚未完全衔接，世应提示内外预期有差异，动爻要求先核对参与条件，再看变化后的风险边界。" * 2 with patch( "backend.features.heaven.agent.llm_transport.chat_completion", side_effect=[ SimpleNamespace(content=invalid, latency_ms=1), SimpleNamespace(content=valid, latency_ms=1), ], ) as completion: result = interpret_heaven( "heart", context, "key", "https://example.invalid", "test-model" ) self.assertEqual(result["answer"], valid) self.assertEqual(completion.call_count, 2) if __name__ == "__main__": unittest.main()