refactor: 统一主聊天系统提示词并接入组装轨迹

app/ai/prompts/agent_loop.py

@@ -1,6 +1,6 @@
 """Agent loop 提示词常量。
 
-当前先以代码常量维护，待搜索/深读策略稳定后迁移到 PromptHub。
+主聊天规则以代码维护；摘要和工具说明保留现有解析路径。
 """
 
 from __future__ import annotations
@@ -16,14 +16,14 @@
 def build_current_date_system_prompt(now: datetime | None = None) -> str:
     """为 LLM 注入当前真实日期，避免模型凭训练 cutoff 猜年份。"""
     current = now or datetime.now(CHINA_TZ)
+    current = current.replace(tzinfo=CHINA_TZ) if current.tzinfo is None else current.astimezone(CHINA_TZ)
     weekday_cn = ["一", "二", "三", "四", "五", "六", "日"][current.weekday()]
     tomorrow = current.date() + timedelta(days=1)
     week_start = current.date() - timedelta(days=current.weekday())
     this_saturday = week_start + timedelta(days=5)
     this_sunday = week_start + timedelta(days=6)
     next_saturday = this_saturday + timedelta(days=7)
     next_sunday = this_sunday + timedelta(days=7)
-    forbidden_years = ", ".join(str(y) for y in range(current.year - 3, current.year))
     return (
         f"【当前真实日期】{current.year}年{current.month}月{current.day}日（星期{weekday_cn}），"
         f"北京时间 {current.strftime('%H:%M')}。\n\n"
@@ -35,8 +35,7 @@ def build_current_date_system_prompt(now: datetime | None = None) -> str:
         f"下周日是 {_format_date_with_weekday(next_sunday)}。\n\n"
         f"硬性规则：\n"
         f"1. 涉及『最新』『当前』『目前』等时效性问题，必须基于上述日期作答。\n"
-        f"2. **生成搜索关键词时，年份必须使用 {current.year} 或更晚，"
-        f"严禁使用 {forbidden_years} 等过去年份**。\n"
+        f"2. 搜索年份应匹配用户问题；历史事件或指定年份查询应保留对应历史年份，不得强制改为当前年份。\n"
         f"3. 不要相信训练数据中的『当前年份』印象——你的训练 cutoff 早于现在。\n"
         f"4. 用户使用『今天』『明天』『本周末』『下周末』等相对日期时，"
         f"必须先按上述锚点换算；搜索词与最终答案中的日期、星期必须一致。"
@@ -262,25 +261,25 @@ def get_runtime_prompt_template(name: str, fallback: str) -> str:
 
 
 def get_app_identity_prompt() -> str:
-    return get_runtime_prompt_template("app_identity", APP_IDENTITY_PROMPT)
+    return APP_IDENTITY_PROMPT
 
 
 def get_tool_usage_contract_prompt() -> str:
-    return get_runtime_prompt_template("tool_usage_contract", TOOL_USAGE_CONTRACT_PROMPT)
+    return TOOL_USAGE_CONTRACT_PROMPT
 
 
 def get_no_tool_network_boundary_prompt() -> str:
-    return get_runtime_prompt_template("no_tool_network_boundary", NO_TOOL_NETWORK_BOUNDARY_PROMPT)
+    return NO_TOOL_NETWORK_BOUNDARY_PROMPT
 
 
 def get_agent_plan_control_prompt(plan_mode: str) -> str:
     if plan_mode == "on":
-        return get_runtime_prompt_template("agent_plan_control_on", AGENT_PLAN_CONTROL_ON_PROMPT)
-    return get_runtime_prompt_template("agent_plan_control_auto", AGENT_PLAN_CONTROL_AUTO_PROMPT)
+        return AGENT_PLAN_CONTROL_ON_PROMPT
+    return AGENT_PLAN_CONTROL_AUTO_PROMPT
 
 
 def get_no_vision_file_boundary_prompt() -> str:
-    return get_runtime_prompt_template("no_vision_file_boundary", NO_VISION_FILE_BOUNDARY_PROMPT)
+    return NO_VISION_FILE_BOUNDARY_PROMPT
 
 
 def get_url_read_tool_description() -> str:
@@ -292,4 +291,4 @@ def get_limit_summary_prompt() -> str:
 
 
 def get_continuation_system_prompt() -> str:
-    return get_runtime_prompt_template("continuation_system", CONTINUATION_SYSTEM_PROMPT)
+    return CONTINUATION_SYSTEM_PROMPT

app/ai/prompts/system_prompt.py

@@ -0,0 +1,89 @@
+"""按可信段落身份组装系统提示词；仅处理内存数据，不访问服务或外部 IO。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass
+from time import perf_counter
+from typing import Any
+
+from app.ai.prompts.agent_loop import build_current_date_system_prompt, get_app_identity_prompt
+from app.utils.prompt_fingerprint import fingerprint_system_messages
+
+TEMPLATE_VERSION = "2026-08-26.1"
+
+
+@dataclass(frozen=True)
+class SystemPromptSection:
+    section_id: str
+    content: str
+
+
+@dataclass(frozen=True)
+class SystemPromptAssembly:
+    messages: list[dict]
+    metadata: dict[str, Any]
+
+
+class SystemPromptAssemblyError(Exception):
+    """仅向调用方提供固定安全错误信息和组装元数据。"""
+
+    def __init__(self, metadata: dict[str, Any]):
+        super().__init__("系统提示词组装失败")
+        self.metadata = metadata
+
+
+def build_base_sections(user_system_prompt: str | None = None) -> list[SystemPromptSection]:
+    sections = [SystemPromptSection("current_date", build_current_date_system_prompt())]
+    if user_system_prompt and user_system_prompt.strip():
+        sections.append(
+            SystemPromptSection(
+                "user_preferences",
+                "以下是用户的个性化偏好设置，请在回答中自然遵守，但不要主动提及这些设置本身：\n\n"
+                + user_system_prompt.strip(),
+            )
+        )
+    sections.append(SystemPromptSection("app_identity", get_app_identity_prompt()))
+    return sections
+
+
+def assemble_system_prompt(
+    *,
+    user_system_prompt: str | None = None,
+    sections: Callable[[], Iterable[SystemPromptSection]] | None = None,
+) -> SystemPromptAssembly:
+    """回调只允许读取已准备的内存状态，段落内容不参与身份判定。"""
+    started = perf_counter()
+    section_ids: list[str] = []
+    metadata: dict[str, Any] = {
+        "status": "failed",
+        "source": "code",
+        "template_version": TEMPLATE_VERSION,
+        "section_ids": section_ids,
+        "fingerprint": None,
+        "char_count": None,
+    }
+    try:
+        selected = build_base_sections(user_system_prompt)
+        if sections is not None:
+            selected.extend(sections())
+        messages = []
+        for section in selected:
+            if section.section_id in section_ids:
+                continue
+            section_ids.append(section.section_id)
+            messages.append({"role": "system", "content": section.content})
+        metadata.update(
+            status="ready",
+            fingerprint=fingerprint_system_messages(messages),
+            char_count=sum(len(message["content"]) for message in messages),
+            duration_ms=max(0, int((perf_counter() - started) * 1000)),
+        )
+        return SystemPromptAssembly(messages=messages, metadata=metadata)
+    except Exception:
+        metadata.update(
+            duration_ms=max(0, int((perf_counter() - started) * 1000)),
+            error_code="assembly_failed",
+            message="系统提示词组装失败",
+        )
+        raise SystemPromptAssemblyError(metadata) from None

app/services/agent/emitter.py

@@ -271,6 +271,7 @@ async def llm_round_started(
         model: str,
         provider: str,
         parent_step_id: str | None = None,
+        system_prompt_fingerprint: str | None = None,
     ) -> None:
         await self._emit(
             ev.LLMRoundStarted(
@@ -279,6 +280,7 @@ async def llm_round_started(
                 round_index=round_index,
                 model=model,
                 provider=provider,
+                system_prompt_fingerprint=system_prompt_fingerprint,
                 **self._envelope(parent_step_id=parent_step_id),
             )
         )
@@ -636,6 +638,40 @@ async def content_block_discarded(self, *, block_id: str) -> None:
             )
         )
 
+    async def system_prompt_prepared(
+        self,
+        *,
+        status: str,
+        source: str,
+        template_version: str,
+        section_ids: list[str],
+        duration_ms: int,
+        fingerprint: str | None = None,
+        char_count: int | None = None,
+        error_code: str | None = None,
+        message: str | None = None,
+    ) -> None:
+        if self._message_id is None:
+            raise RuntimeError("system_prompt_prepared 必须在 run_started 之后发送")
+        # 错误原文可能含偏好或模板内容；事件只发送固定文案。
+        safe_error = "assembly_failed" if status == "failed" else None
+        await self._emit(
+            ev.SystemPromptPrepared(
+                type="system_prompt_prepared",
+                protocol_version=2,
+                status=status,
+                source=source,
+                template_version=template_version,
+                section_ids=section_ids,
+                fingerprint=fingerprint,
+                char_count=char_count,
+                duration_ms=duration_ms,
+                error_code=safe_error,
+                message="系统提示词组装失败，请稍后重试。" if safe_error else None,
+                **self._envelope(step_id=None),
+            )
+        )
+
     async def context_status_updated(
         self,
         *,

app/services/agent/events.py

@@ -101,6 +101,7 @@ class LLMRoundStarted(AgentEventBase):
     round_index: int = Field(ge=1)
     model: str
     provider: str
+    system_prompt_fingerprint: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
 
 
 class LLMRoundFirstOutputDelta(AgentEventBase):
@@ -301,6 +302,22 @@ class ContentBlockDiscarded(AgentEventBase):
     block_id: str = Field(min_length=1, max_length=128)
 
 
+class SystemPromptPrepared(AgentEventBase):
+    """本地系统提示词组装的终态元数据，不携带提示词或用户偏好正文。"""
+
+    type: Literal["system_prompt_prepared"]
+    protocol_version: Literal[2]
+    status: Literal["ready", "failed"]
+    source: Literal["code"]
+    template_version: str = Field(min_length=1, max_length=64)
+    section_ids: list[str] = Field(max_length=50)
+    fingerprint: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
+    char_count: int | None = Field(default=None, ge=0)
+    duration_ms: int = Field(ge=0)
+    error_code: str | None = None
+    message: str | None = Field(default=None, max_length=120)
+
+
 class ContextStatusUpdated(AgentEventBase):
     """单轮 LLM 上下文状态；字段严格白名单，不携带 prompt 或内部来源。"""
 
@@ -371,6 +388,7 @@ class ContextResult(AgentEventBase):
     | EvidenceItemUpserted
     | ContentBlockUpserted
     | ContentBlockDiscarded
+    | SystemPromptPrepared
     | ContextStatusUpdated
     | ContextRequired
     | ContextResult,

app/services/agent/trajectory_payload.py

@@ -36,7 +36,7 @@
     "run_interrupted": frozenset({"reason"}),
     "run_failed": frozenset({"error_code", "message"}),
     "run_completed": frozenset({"total_steps", "total_tool_calls", "finish_reason"}),
-    "llm_round_started": frozenset({"llm_round_id", "round_index", "model", "provider"}),
+    "llm_round_started": frozenset({"llm_round_id", "round_index", "model", "provider", "system_prompt_fingerprint"}),
     "llm_round_first_output_delta": frozenset({"llm_round_id", "delta_kind", "ttft_ms"}),
     "llm_round_completed": frozenset(
         {
@@ -94,6 +94,20 @@
     # 完整 content block 不属于 P0 脱敏账本，只保留事件存在性与协议版本。
     "content_block_upserted": frozenset({"protocol_version"}),
     "content_block_discarded": frozenset({"protocol_version", "block_id"}),
+    "system_prompt_prepared": frozenset(
+        {
+            "protocol_version",
+            "status",
+            "source",
+            "template_version",
+            "section_ids",
+            "fingerprint",
+            "char_count",
+            "duration_ms",
+            "error_code",
+            "message",
+        }
+    ),
     "context_status_updated": frozenset(
         {
             "protocol_version",
@@ -146,7 +160,7 @@
         "citation_index",
     }
 )
-_LIST_FIELDS = frozenset({"tools", "key_findings", "source_refs"})
+_LIST_FIELDS = frozenset({"tools", "key_findings", "source_refs", "section_ids"})
 _SECRET_PATTERN = re.compile(
     r"(?i)\b(api[_-]?key|authorization|access[_-]?token|token|password|secret)\s*[:=]\s*"
     r"(?:bearer\s+)?[^\s,;]+"

app/services/chat/message_builder.py

@@ -10,7 +10,7 @@
 import base64
 from typing import Dict, List, Optional
 
-from app.ai.prompts.agent_loop import build_current_date_system_prompt, get_app_identity_prompt
+from app.ai.prompts.system_prompt import build_base_sections
 from app.core.logger import app_logger as logger
 from app.db.repositories import FileRepository
 from app.services.file_service import is_image_mime
@@ -66,6 +66,7 @@ async def build_llm_messages(
     *,
     user_id: Optional[str] = None,
     conversation_id: Optional[str] = None,
+    include_base_system: bool = True,
 ) -> List[dict]:
     """
     将 content blocks 消息列表转为 LLM 可消费的 dict 格式。
@@ -79,24 +80,12 @@ async def build_llm_messages(
     """
     result = []
 
-    # 注入当前日期（始终注入，对所有对话生效）
-    result.append({"role": "system", "content": build_current_date_system_prompt()})
-
-    # 注入用户自定义个性化 prompt
-    if user_system_prompt and user_system_prompt.strip():
-        result.append(
-            {
-                "role": "system",
-                "content": (
-                    "以下是用户的个性化偏好设置，请在回答中自然遵守，但不要主动提及这些设置本身：\n\n"
-                    f"{user_system_prompt.strip()}"
-                ),
-            }
+    # 主聊天由纯组装器统一注入；其他调用保持基础规则保证。
+    if include_base_system:
+        result.extend(
+            {"role": "system", "content": section.content} for section in build_base_sections(user_system_prompt)
         )
 
-    # 注入产品身份约束（始终注入，并放在用户偏好之后，避免被个性化设置覆盖）
-    result.append({"role": "system", "content": get_app_identity_prompt()})
-
     # 计算最近 N 轮用户消息的起始索引，用于控制图片注入范围
     vision_cutoff_idx = 0
     if has_vision:
@@ -168,12 +157,13 @@ def inject_file_content(
     file_contents: Dict[str, str],
 ) -> List[dict]:
     """将非图片文件的解析内容注入到最后一条用户消息的文本中"""
-    if not messages:
-        return messages
-
     combined = "\n\n".join(f"文件内容 ({i + 1}):\n{content}" for i, content in enumerate(file_contents.values()))
     enhanced = f"{original_message}\n\n以下是相关文件内容，请结合这些内容回答：\n{combined}"
 
+    # 仅附件的新会话没有文字消息，仍须将解析正文作为用户输入。
+    if not messages:
+        return [{"role": "user", "content": enhanced}]
+
     result = messages.copy()
     result[-1] = {"role": "user", "content": enhanced}
     return result