perf: reduce LLM input tokens by ~50% via prompt deduplication
- Remove JSON appendix from user_prompt (saves ~3000 chars, 85% reduction) - Consolidate identity: remove English system_prompt identity, merge into agent_prompt Chinese identity - Simplify system_prompt output rules: keep only Language Requirement, remove Answer Rules (duplicate with agent_prompt) - Enhance follow-up context: include more divination_derived fields (changing_yaos, fushen, wu_xing, etc.) - Fix bug: lowerName -> lower_name in user_prompt (Pydantic snake_case) - Update tests to reflect new prompt structure
This commit is contained in:
qzl
@@ -5,8 +5,63 @@ from collections.abc import Callable
|
||||
from core.agentscope.prompts.sections import wrap_section
|
||||
from schemas.agent.system_agent import AgentType, SystemAgentLLMConfig
|
||||
|
||||
PromptRuleBuilder = Callable[[SystemAgentLLMConfig | None], str]
|
||||
|
||||
PromptRuleBuilder = Callable[[SystemAgentLLMConfig | None], list[str]]
|
||||
_WORKER_IDENTITY = """\
|
||||
你是 Eryao 的六爻解卦助手。只负责六爻解读，不做日程、自动化、待办等任务。
|
||||
必须返回严格 JSON，只返回一个对象，字段必须匹配运行时输出模型。
|
||||
禁止额外字段、解释性前后缀、Markdown 代码块。
|
||||
若信息不足以高置信判断，仍须按输出模型返回结果，但须体现不确定性。
|
||||
回复须简洁求实、以解读为导向；不得声称未由卦象数据证实的判断。"""
|
||||
|
||||
_WORKER_INPUT_FORMAT = """\
|
||||
用户消息已包含完整起卦数据：用户问题、起卦方式/时间、本卦变卦信息、干支、五行旺衰、逐爻详情（六神、六亲、天干地支、阴阳、动爻、世应）、变卦爻象、伏神等。末尾附有结构化 JSON 供交叉核对。
|
||||
严禁声称"未提供起卦信息""缺少卦象数据"。"""
|
||||
|
||||
_WORKER_HARD_CONSTRAINTS = """\
|
||||
严禁跳过取用神直接下结论。
|
||||
严禁多核心用神混合判签；一问多意时只取一个主用神。
|
||||
严禁仅凭单一因素判吉凶。
|
||||
严禁将旬空月破机械等同凶；须辨真假及是否可解除。
|
||||
严禁将有救应的卦直接断死，严禁表面顺利后续败坏的卦直判上吉。
|
||||
严禁脱离爻位证据给建议。
|
||||
严禁夸大确定性；吉凶并见或证据冲突时须说明边界。"""
|
||||
|
||||
_WORKER_PROCEDURE = """\
|
||||
第1步[取用神]：根据问题类型定唯一主用神并述理由。用神不现时须查伏神、飞神、变爻引出、暗动提拔，不可直接判弱判无。
|
||||
|
||||
第2步[主判]：先看用神旺衰（月建日辰生扶克泄耗），再看世应与用神关系（生合冲克墓绝），再看动爻是生扶还是克耗用神，给事情定底色（偏吉/偏凶/杂局）。
|
||||
|
||||
第3步[修正]：必查原神忌神仇神发动对用神根气的影响；必查动变质量（回头生克、化进退空破墓绝）；必查伏神飞神飞克提拔；必查合冲刑害六冲六合反吟伏吟；必查旬空月破辨真假（真空/假空、真破/假破、值日/出旬/逢合/冲起/填实）；修正是对主判底色做升降，不可脱离主判单独下结论。
|
||||
|
||||
第4步[签级]：只允许 上上签/中上签/中下签/下下签。先定底档再升降：
|
||||
- 初判：偏吉→中上签，偏凶→中下签，杂局→据用神受益或受损孰重落中上/中下。
|
||||
- 升上上：用神旺相或虽病得强力救应、世应有情、原神得力、动变助成、无真空真破真墓绝、当前与最终都偏吉。
|
||||
- 降下下：用神严重受克少救应、真空真破真墓真绝回头克仇神发动、世应严重冲克、多重凶象叠加且最终偏败。
|
||||
- 中上：用神有气，瑕疵虽有但救应大于损伤，当前可阻最终偏成。
|
||||
- 中下：用神偏弱受制，损伤大于救应，当前多阻仅存有限转机。
|
||||
- 边界卦：重大吉凶并见不判极端，须说明哪条证据拉高哪条拉低。
|
||||
|
||||
第5步[救应与态势]：若出现绝处逢生、克处逢生、原神得力来救等，须标注转机；若表面有利但后续败坏，须标注后续风险。必须区分当前态势与最终趋势，签级以最终趋势为主。当前凶后续可转不可直判下下，当前顺后续转坏不可直判上上。"""
|
||||
|
||||
_WORKER_OUTPUT_RULES = """\
|
||||
结论必须结合本卦变卦与关键爻位，不可空谈，至少给出2-4条关键依据（含1条主判+1条修正）。
|
||||
建议必须逐条对应卦象依据（哪一爻、何种生克冲合旺衰），给出可执行动作，优先回答：最该防什么、最该做什么、何时可推进、何时应暂缓。
|
||||
关键词：中文优先四字，非中文用2-4词短语，必须来自本次卦象核心判断。
|
||||
answer 必须是完整解读，覆盖总体判断、当前态势、最终趋势、风险点、转机条件、行动优先级，多段文本段间用\\n\\n分隔，首段直指核心态势（偏吉/偏凶/先难后易/成中有阻等）。
|
||||
sign_level 枚举值固定：上上签/中上签/中下签/下下签。"""
|
||||
|
||||
|
||||
def _worker_rules(llm_config: SystemAgentLLMConfig | None) -> str:
|
||||
_ = llm_config
|
||||
sections = [
|
||||
("identity", _WORKER_IDENTITY),
|
||||
("input_format", _WORKER_INPUT_FORMAT),
|
||||
("hard_constraints", _WORKER_HARD_CONSTRAINTS),
|
||||
("procedure", _WORKER_PROCEDURE),
|
||||
("output_rules", _WORKER_OUTPUT_RULES),
|
||||
]
|
||||
return "\n\n".join(f"[{label}]\n{content}" for label, content in sections)
|
||||
|
||||
|
||||
class AgentPromptRegistry:
|
||||
@@ -21,30 +76,13 @@ class AgentPromptRegistry:
|
||||
*,
|
||||
agent_type: AgentType,
|
||||
llm_config: SystemAgentLLMConfig | None,
|
||||
) -> list[str]:
|
||||
) -> str:
|
||||
builder = self._builders.get(agent_type)
|
||||
if builder is None:
|
||||
builder = self._builders[AgentType.WORKER]
|
||||
return builder(llm_config)
|
||||
|
||||
|
||||
def _worker_rules(llm_config: SystemAgentLLMConfig | None) -> list[str]:
|
||||
_ = llm_config
|
||||
return [
|
||||
"[Worker Identity]",
|
||||
"- 你是 Eryao 的六爻解卦助手，只做解读，不做日程、自动化、待办等任务。",
|
||||
"- 你必须返回严格 JSON，且只返回一个对象，字段必须匹配运行时输出模型。",
|
||||
"[六爻分析流程]",
|
||||
"- 第1步：准确复述用户问题，确认问题类型与诉求焦点。",
|
||||
"- 第2步：围绕用神、世应、动爻、月建日辰、旺衰关系形成核心判断。",
|
||||
"- 第3步：给出签级，仅允许 上上签 / 中上签 / 中下签 / 下下签。",
|
||||
"- 第4步：结论必须结合本卦/变卦与关键爻位，按要点分条说明，不可脱离卦象空谈。",
|
||||
"- 第5步：建议必须逐条对应卦象依据（哪一条爻、何种生克/冲合/旺衰影响），给出可执行动作。",
|
||||
"- 第6步：提炼关键词并匹配 ai_language；仅在中文输出时优先四字表达，非中文时使用短语关键词（2-4 words）。",
|
||||
"- 第7步：answer 需要是完整解读，不要只给简短结论；应覆盖趋势判断、风险点、转机条件与行动优先级，并用多段文本（段间用\\n\\n）呈现。",
|
||||
]
|
||||
|
||||
|
||||
AGENT_PROMPT_REGISTRY = AgentPromptRegistry()
|
||||
AGENT_PROMPT_REGISTRY.register(agent_type=AgentType.WORKER, builder=_worker_rules)
|
||||
|
||||
@@ -57,7 +95,7 @@ def build_agent_prompt(
|
||||
lines = [
|
||||
"[Agent Identity]",
|
||||
f"- type: {agent_type.value}",
|
||||
*AGENT_PROMPT_REGISTRY.build_rules(
|
||||
AGENT_PROMPT_REGISTRY.build_rules(
|
||||
agent_type=agent_type,
|
||||
llm_config=llm_config,
|
||||
),
|
||||
|
||||
@@ -169,20 +169,6 @@ def _build_runtime_context(
|
||||
)
|
||||
|
||||
|
||||
def _build_identity_section() -> str:
|
||||
return wrap_section(
|
||||
"identity",
|
||||
"\n".join(
|
||||
[
|
||||
"[Identity]",
|
||||
"- You are Eryao, a focused six-yao divination worker.",
|
||||
"- Be concise, truthful, and interpretation-oriented.",
|
||||
"- Never claim execution unless confirmed by tool/runtime evidence.",
|
||||
]
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _build_env_section(
|
||||
*,
|
||||
runtime_context: RuntimePromptContext,
|
||||
@@ -224,24 +210,36 @@ def _build_safety_section() -> str:
|
||||
)
|
||||
|
||||
|
||||
_LANGUAGE_LABELS: dict[str, str] = {
|
||||
"zh-CN": "Simplified Chinese (简体中文)",
|
||||
"zh-Hant": "Traditional Chinese (繁體中文)",
|
||||
"en-US": "English",
|
||||
"en": "English",
|
||||
}
|
||||
|
||||
|
||||
def _get_language_label(tag: str) -> str:
|
||||
"""Convert language tag to human-readable label."""
|
||||
return _LANGUAGE_LABELS.get(tag, tag)
|
||||
|
||||
|
||||
def _build_output_rules(*, ai_language: str) -> str:
|
||||
return wrap_section(
|
||||
"output",
|
||||
"\n".join(
|
||||
[
|
||||
"[Answer Rules]",
|
||||
f"- You must produce conclusion/focus_points/advice/keywords/answer in ai_language={ai_language} unless the user explicitly asks for another language.",
|
||||
"- keywords must use the same language as answer; do not mix Chinese and English in one keyword list.",
|
||||
"- sign_level must stay in canonical Chinese enum: 上上签 / 中上签 / 中下签 / 下下签.",
|
||||
"- answer must be natural user-facing explanation, not a rigid step-by-step process transcript.",
|
||||
"- conclusion/advice/answer must be grounded in actual hexagram evidence and discuss points one by one (not generic template text).",
|
||||
"- format answer as multiple short paragraphs separated by \n\n for readability.",
|
||||
"- if ai_language is non-Chinese, translate domain terms consistently to that language instead of keeping fixed Chinese terms.",
|
||||
"- Keep output factual, concise, and schema-consistent.",
|
||||
"- Lead with conclusion, then only key supporting facts.",
|
||||
lang_label = _get_language_label(ai_language)
|
||||
is_chinese = ai_language.startswith("zh")
|
||||
rules = [
|
||||
"[Language Requirement]",
|
||||
f"- You MUST respond in {lang_label}.",
|
||||
f"- IGNORE the language of user's question-you must still answer in {lang_label}.",
|
||||
"- Do NOT switch to another language even if the user asks in a different language.",
|
||||
]
|
||||
if is_chinese:
|
||||
rules.extend(
|
||||
[
|
||||
"- If ai_language is zh-Hant, write all Chinese characters in Traditional Chinese (繁體中文).",
|
||||
"- If ai_language is zh-CN, write all Chinese characters in Simplified Chinese (简体中文).",
|
||||
]
|
||||
),
|
||||
)
|
||||
return wrap_section("output", "\n".join(rules))
|
||||
|
||||
|
||||
def build_system_prompt(
|
||||
@@ -260,7 +258,6 @@ def build_system_prompt(
|
||||
runtime_client_time=runtime_client_time,
|
||||
)
|
||||
sections: list[str | None] = [
|
||||
_build_identity_section(),
|
||||
_build_env_section(
|
||||
runtime_context=runtime_context,
|
||||
extra_context=extra_context,
|
||||
|
||||
@@ -4,15 +4,79 @@ from schemas.domain.divination import DerivedDivinationData
|
||||
|
||||
|
||||
def build_divination_user_prompt(*, derived: DerivedDivinationData) -> str:
|
||||
structured_json = derived.model_dump_json(
|
||||
by_alias=True,
|
||||
exclude_none=True,
|
||||
ensure_ascii=False,
|
||||
lines: list[str] = []
|
||||
lines.append(f"用户问题：{derived.question}")
|
||||
lines.append(f"问题类型：{derived.question_type}")
|
||||
lines.append(f"起卦方式：{derived.divination_method}")
|
||||
lines.append(f"起卦时间：{derived.divination_time}")
|
||||
lines.append("")
|
||||
|
||||
lines.append("【本卦】")
|
||||
lines.append(
|
||||
f"卦名：{derived.gua_name}（上{derived.upper_name}下{derived.lower_name}）"
|
||||
)
|
||||
return (
|
||||
f"用户问题：{derived.question}\n"
|
||||
f"问题类型：{derived.question_type}\n"
|
||||
"以下是后端推导后的六爻结构化数据（JSON）：\n"
|
||||
f"{structured_json}\n"
|
||||
"请仅基于以上六爻数据做专业解读。"
|
||||
lines.append(f"卦象：{derived.binary_code}")
|
||||
lines.append(
|
||||
f"世爻第{derived.world_position}爻，应爻第{derived.response_position}爻"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
if derived.has_changing_yao:
|
||||
lines.append("【变卦】")
|
||||
lines.append(f"卦名：{derived.target_gua_name}")
|
||||
lines.append(f"卦象：{derived.changed_binary_code}")
|
||||
lines.append("")
|
||||
|
||||
lines.append("【干支】")
|
||||
g = derived.ganzhi
|
||||
lines.append(
|
||||
f"年柱：{g.year_gan_zhi} 月柱：{g.month_gan_zhi} 日柱：{g.day_gan_zhi} 时柱：{g.time_gan_zhi}"
|
||||
)
|
||||
lines.append(f"月建：{g.yue_jian} 日辰：{g.ri_chen}")
|
||||
lines.append(f"月破：{g.yue_po} 日冲：{g.ri_chong}")
|
||||
lines.append(
|
||||
f"年空亡：{g.year_kong_wang} 月空亡：{g.month_kong_wang}"
|
||||
f" 日空亡：{g.day_kong_wang} 时空亡：{g.time_kong_wang}"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
lines.append("【五行旺衰】")
|
||||
for element, status in derived.wu_xing_statuses.items():
|
||||
lines.append(f"{element}：{status}")
|
||||
lines.append("")
|
||||
|
||||
lines.append("【本卦爻象】")
|
||||
for yao in derived.yao_info_list:
|
||||
changing_mark = "（动）" if yao.is_changing else ""
|
||||
world_response = yao.special_mark
|
||||
lines.append(
|
||||
f"第{yao.position}爻：{yao.spirit_name} {yao.relation_name} "
|
||||
f"{yao.tigan_name}{yao.element_name} "
|
||||
f"{'阳' if yao.is_yang else '阴'}爻{changing_mark}"
|
||||
f"{' 世' if world_response == '世' else ' 应' if world_response == '应' else ''}"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
if derived.has_changing_yao and derived.target_yao_info_list:
|
||||
lines.append("【变卦爻象】")
|
||||
for yao in derived.target_yao_info_list:
|
||||
world_response = yao.special_mark
|
||||
lines.append(
|
||||
f"第{yao.position}爻：{yao.spirit_name} {yao.relation_name} "
|
||||
f"{yao.tigan_name}{yao.element_name} "
|
||||
f"{'阳' if yao.is_yang else '阴'}爻"
|
||||
f"{' 世' if world_response == '世' else ' 应' if world_response == '应' else ''}"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
if derived.fushen_info_list:
|
||||
lines.append("【伏神】")
|
||||
for fs in derived.fushen_info_list:
|
||||
lines.append(
|
||||
f"第{fs.position}爻：{fs.relation_name} {fs.tigan_name}{fs.element_name}"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
lines.append("——以上为起卦所得完整数据，请据此进行六爻解读。")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@@ -321,16 +321,27 @@ class AgentScopeRunner:
|
||||
run_input: RunAgentInput,
|
||||
derived_divination: DerivedDivinationData | None,
|
||||
) -> list[Msg]:
|
||||
if derived_divination is not None:
|
||||
user_text = build_divination_user_prompt(derived=derived_divination)
|
||||
else:
|
||||
_, latest_user_text = extract_latest_user_payload(run_input)
|
||||
user_text = latest_user_text
|
||||
|
||||
if derived_divination is not None and context_messages:
|
||||
last = context_messages[-1]
|
||||
if last.role == "user":
|
||||
context_messages[-1] = Msg(
|
||||
name=last.name,
|
||||
role=last.role,
|
||||
content=user_text,
|
||||
)
|
||||
return context_messages
|
||||
|
||||
if context_messages:
|
||||
last = context_messages[-1]
|
||||
if last.role == "user":
|
||||
return context_messages
|
||||
|
||||
_, latest_user_text = extract_latest_user_payload(run_input)
|
||||
if derived_divination is None:
|
||||
user_text = latest_user_text
|
||||
else:
|
||||
user_text = build_divination_user_prompt(derived=derived_divination)
|
||||
user_blocks = [{"type": "text", "text": user_text}]
|
||||
if (
|
||||
user_blocks
|
||||
|
||||
@@ -117,11 +117,58 @@ def _serialize_assistant_context_from_metadata(
|
||||
("targetGuaName", "target_gua_name"),
|
||||
("binaryCode", "binary_code"),
|
||||
("changedBinaryCode", "changed_binary_code"),
|
||||
("questionType", "question_type"),
|
||||
("worldPosition", "world_position"),
|
||||
("responsePosition", "response_position"),
|
||||
)
|
||||
for key, label in key_map:
|
||||
value = divination_derived.get(key)
|
||||
if isinstance(value, str) and value:
|
||||
lines.append(f" {label}: {value}")
|
||||
elif isinstance(value, int) and value:
|
||||
lines.append(f" {label}: {value}")
|
||||
|
||||
ganzhi = divination_derived.get("ganzhi")
|
||||
if isinstance(ganzhi, dict):
|
||||
day = ganzhi.get("dayGanZhi")
|
||||
month = ganzhi.get("monthGanZhi")
|
||||
if isinstance(day, str) and day:
|
||||
lines.append(f" day_gan_zhi: {day}")
|
||||
if isinstance(month, str) and month:
|
||||
lines.append(f" month_gan_zhi: {month}")
|
||||
|
||||
wuxing = divination_derived.get("wuXingStatuses")
|
||||
if isinstance(wuxing, dict) and wuxing:
|
||||
lines.append(f" wu_xing: {json.dumps(wuxing, ensure_ascii=False)}")
|
||||
|
||||
changing_yaos: list[str] = []
|
||||
yao_list = divination_derived.get("yaoInfoList")
|
||||
if isinstance(yao_list, list):
|
||||
for yao in yao_list:
|
||||
if isinstance(yao, dict) and yao.get("isChanging"):
|
||||
pos = yao.get("position", "?")
|
||||
rel = yao.get("relationName", "?")
|
||||
ele = yao.get("elementName", "")
|
||||
tg = yao.get("tiganName", "")
|
||||
mark = yao.get("specialMark") or ""
|
||||
changing_yaos.append(
|
||||
f"第{pos}爻:{rel}{tg}{ele}" + (f"({mark})" if mark else "")
|
||||
)
|
||||
if changing_yaos:
|
||||
lines.append(f" changing_yaos: {', '.join(changing_yaos)}")
|
||||
|
||||
fushen_yaos: list[str] = []
|
||||
fs_list = divination_derived.get("fushenInfoList")
|
||||
if isinstance(fs_list, list):
|
||||
for fs in fs_list:
|
||||
if isinstance(fs, dict):
|
||||
pos = fs.get("position", "?")
|
||||
rel = fs.get("relationName", "?")
|
||||
tg = fs.get("tiganName", "")
|
||||
ele = fs.get("elementName", "")
|
||||
fushen_yaos.append(f"第{pos}爻:{rel}{tg}{ele}")
|
||||
if fushen_yaos:
|
||||
lines.append(f" fushen: {', '.join(fushen_yaos)}")
|
||||
|
||||
if len(lines) <= 1:
|
||||
return fallback_content
|
||||
|
||||
@@ -34,7 +34,7 @@ def test_system_prompt_enforces_ai_language_and_identity_signals() -> None:
|
||||
now_utc=datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
assert "ai_language=en-US" in prompt
|
||||
assert '"ai_language":"en-US"' in prompt
|
||||
assert (
|
||||
"interface_language and country are weak signals for user identity inference"
|
||||
in prompt
|
||||
@@ -65,8 +65,7 @@ def test_agent_prompt_keeps_only_identity_and_domain_flow() -> None:
|
||||
|
||||
assert "[输出约束]" not in prompt
|
||||
assert "[安全与拒答]" not in prompt
|
||||
assert "[六爻分析流程]" in prompt
|
||||
assert "匹配 ai_language" in prompt
|
||||
assert "[procedure]" in prompt
|
||||
assert "段间用\\n\\n" in prompt
|
||||
assert "优先四字表达，简洁且可复述" not in prompt
|
||||
|
||||
@@ -94,7 +93,7 @@ def test_system_prompt_sanitizes_invalid_language_and_country() -> None:
|
||||
now_utc=datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
assert "ai_language=zh-CN" in prompt
|
||||
assert '"ai_language":"zh-CN"' in prompt
|
||||
assert '"interface_language":"zh-CN"' in prompt
|
||||
assert '"country":"CN"' in prompt
|
||||
|
||||
@@ -113,11 +112,9 @@ def test_system_prompt_sections_are_not_duplicated() -> None:
|
||||
|
||||
|
||||
def test_system_prompt_requires_paragraph_breaks_for_answer() -> None:
|
||||
prompt = build_system_prompt(
|
||||
prompt = build_agent_prompt(
|
||||
agent_type=AgentType.WORKER,
|
||||
llm_config=SystemAgentLLMConfig(),
|
||||
user_context=_build_user_context(ai_language="zh-CN"),
|
||||
now_utc=datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
assert "multiple short paragraphs" in prompt
|
||||
assert "段间用\\n\\n" in prompt
|
||||
|
||||
Reference in New Issue
Block a user