refactor(prompt): Prompt 集中化 - 新增 fam-edge ai_orchestrator/prompts.py（视频分析/智能问答/人物合并三模板共享，消除 gemini/nvidia 两份发散）；gemini/nvidia _build_video_prompt 改调共享函数并注入 camera_name；person_service 改调 build_person_merge_prompt；fam-core chat_handler 独立 prompts.py（跨模块风格一致）替代内联 CHAT_SYSTEM_PROMPT；输出硬约束(首字符{/禁markdown)、3 秒密度抽取、people_mentioned 一致性强制、描述 7 维度、人物合并唯一性约束

fam-core/src/fam_core/chat_handler/chat_handler.py

@@ -18,28 +18,12 @@ from flask import Blueprint, request, jsonify
 from ..logger import setup_logger
 from ..config_loader import load_config
 from .. import db_layer
+from .prompts import build_chat_prompt
 
 logger = setup_logger('fam-core.chat_handler')
 
 chat_bp = Blueprint('chat_handler', __name__)
 
-CHAT_SYSTEM_PROMPT = """你是家庭监控助手。根据以下监控数据，回答用户问题。
-
-监控数据（按时间顺序，每条一行）：
-{context}
-
-已知家庭成员: {members}
-
-用户问题: {question}
-
-要求:
-- 只基于上述数据回答，不要编造
-- 按时间顺序总结
-- 若有关注事件（跌倒、哭闹、陌生人等），重点提示
-- 若当天没有该人员的数据，明确说"今天没有观察到{person}"
-- 用自然语言回答，不要输出 JSON
-"""
-
 
 def _format_events(rows):
     """将 sync_events 查询行格式化为上下文文本"""
@@ -104,11 +88,11 @@ def chat_ask():
         context = _format_events(rows)
         context_summary = f"查询 sync_events {len(rows)} 条"
         members = db_layer.get_sync_known_members_context()
-        prompt = CHAT_SYSTEM_PROMPT.format(
+        prompt = build_chat_prompt(
             context=context,
             members=members or queried_person,
             question=question,
-            person=queried_person
+            queried_person=queried_person
         )
         try:
             answer = _call_edge_qa(prompt)

fam-core/src/fam_core/chat_handler/prompts.py

@@ -0,0 +1,38 @@
+"""Chat-Handler 的 Prompt 模板。
+
+fam-core 不能 import fam-edge 的 ai_orchestrator，故 chat prompt 独立维护于此。
+设计原则见 fam-edge/src/fam_edge/ai_orchestrator/prompts.py（跨模块不共享，但风格一致）。
+"""
+
+
+def build_chat_prompt(context: str, members: str, question: str,
+                      queried_person: str) -> str:
+    """构建智能问答 system prompt。
+
+    Args:
+      context: sync_events 格式化后的上下文文本（每行一条事件）
+      members: known_members_context 文本
+      question: 用户原始问题
+      queried_person: 用户指定的人物
+    """
+    members_block = members.strip() if members and members.strip() else queried_person
+    context_block = context.strip() if context and context.strip() else "（今日无该人员的监控记录）"
+
+    return f"""你是家庭监控助手。仅根据下方监控数据回答用户问题。
+
+【监控数据】（按时间顺序，每行一条事件）
+{context_block}
+
+【已知家庭成员】
+{members_block}
+
+【用户问题】
+{question}
+（用户关心的人物：{queried_person}）
+
+【回答要求】
+1. 只基于上述监控数据，不编造、不补充数据外信息。
+2. 按时间顺序组织回答，突出关键事件。
+3. 若有关注事件（跌倒、哭闹、陌生人等），重点提示。
+4. 若数据为空或当天未观察到 {queried_person}，明确说"今天没有观察到{queried_person}"。
+5. 用自然语言回答，不要输出 JSON 或列表格式。"""

fam-edge/src/fam_edge/ai_orchestrator/__init__.py

@@ -1,4 +1,8 @@
 """AI-Orchestrator 包"""
 from .json_parser import parse_vlm_json, validate_schema, VLMOutputInvalidError
+from .prompts import build_video_prompt, build_chat_prompt, build_person_merge_prompt
 
-__all__ = ["parse_vlm_json", "validate_schema", "VLMOutputInvalidError"]
+__all__ = [
+    "parse_vlm_json", "validate_schema", "VLMOutputInvalidError",
+    "build_video_prompt", "build_chat_prompt", "build_person_merge_prompt",
+]

fam-edge/src/fam_edge/ai_orchestrator/prompts.py

@@ -0,0 +1,143 @@
+"""
+Prompt 模板 - 集中管理，避免 gemini/nvidia 适配器各维护一份导致发散。
+
+设计原则:
+  1. schema 用真实 JSON 示例展示，不靠文字描述字段名（VLM 对示例比对纯文字更可靠）
+  2. timestamp 统一"视频内相对时间 HH:MM:SS"，消除绝对/相对歧义
+  3. 人物命名: 已知成员用真名，未知用"人物A/B/C"本视频内临时编号，
+     并强制 people_mentioned = events 内出现人物去重后的集合（下游合并依赖）
+  4. 输出硬约束: 首字符必须是 {，禁止思考过程/markdown/解释
+  5. 边界情况: 无人/空视频/看不清 -> 空 events + summary 说明，不凑数
+"""
+from typing import Optional
+
+
+def build_video_prompt(known_members: str, event_start_time: str,
+                       camera_name: str = '') -> str:
+    """构建整视频分析 prompt（gemini/nvidia 共用）。
+
+    Args:
+      known_members: get_known_members_context() 输出，每行 "- 真名（别名/标识：label）" 或 "- label"
+      event_start_time: 从文件名解析的视频开始时间（北京时间），仅用于 start_hint
+      camera_name: 摄像头名（注入提示，帮助模型理解画面位置语境）
+    """
+    camera_hint = f"\n摄像头位置：{camera_name}。" if camera_name else ""
+    start_hint = ""
+    if event_start_time:
+        start_hint = (f"\n视频开始时间（北京时间）约为 {event_start_time}，"
+                      f"但 timestamp 字段仍填视频内相对时间（见下方格式说明）。")
+
+    members_block = known_members.strip() if known_members and known_members.strip() else "（暂无已知成员，所有人物用「人物A」「人物B」编号）"
+
+    return f"""你是家庭监控视频分析助手。请观看整段监控录像，提取结构化信息。
+
+【输出格式 - 必须严格遵守】
+- 只输出一个合法 JSON 对象，首字符必须是 {{，末字符必须是 }}。
+- 禁止输出 markdown 代码块、思考过程、解释文字、前后缀。
+- 空视频/画面看不清/无人出现时，events 填空数组，global_summary 说明情况。
+
+【JSON 结构】
+{{
+  "global_summary": "整个时段的客观摘要，简体中文，2-4 句，说明谁在做什么",
+  "events": [
+    {{
+      "timestamp": "HH:MM:SS",
+      "description": "该时刻画面的详细描述：人物身份、动作细节、位置移动、交互对象、姿态/手势/朝向、手中物品、周围环境",
+      "people": ["人物标识"],
+      "is_attention_event": false
+    }}
+  ],
+  "people_mentioned": ["本视频出现的所有人物标识去重后的集合"]
+}}
+
+【字段规则】
+1. timestamp: 视频内相对时间，格式 HH:MM:SS，从视频开头 00:00:00 算起。
+   示例：视频开始后 5 分 23 秒 → "00:05:23"。禁止输出绝对日期时间。
+2. events 抽取密度（核心规则）:
+   a) 有人出现在画面中时，每隔约 3 秒抽取一帧作为一个 event，时间戳对齐到 3 的倍数
+      （如 00:00:00、00:00:03、00:00:06、00:00:09……）。
+      示例：某人从 00:00:05 走入画面，00:00:30 离开 → 生成 00:00:06、00:00:09、
+      00:00:12、……、00:00:27 共约 8 条 event，每条描述该 3 秒窗口内的动作变化。
+   b) 同一人物持续在画面中且动作无明显变化时，仍按 3 秒一帧抽取，但 description 须描述
+      该 3 秒内的细微变化（姿态、位置、朝向、与谁交谈等），不要简单重复上一条。
+   c) 人物进入/离开画面的瞬间必须各成一条 event（时间戳取实际发生时刻，不必对齐 3 秒）。
+   d) 动作发生显著变化（如从走动变为坐下、从哭泣变为安静、拿取物品）的转折点必须成一条 event。
+   e) 无人出现的时段不要单独成 event。
+3. description 必须尽量详细，每条至少覆盖以下维度（缺失的维度写"无"）:
+   - 人物身份（用真名或「人物A」编号）+ 当前动作（走动/站立/坐下/蹲下/弯腰/奔跑等）
+   - 位置（如"客厅沙发左侧"/"厨房门口"/"走廊中部"）+ 移动方向（向门口走/原地不动/朝镜头靠近）
+   - 姿态（站姿/蹲姿/坐姿）+ 朝向（面向镜头/背对镜头/侧身）
+   - 手部动作（手里拿着杯子/双手插兜/挥手/扶墙/抱孩子等）
+   - 交互对象（与谁交谈/喂食/搀扶/推搡/独处）
+   - 表情/情绪线索（如可辨认：微笑/皱眉/哭泣/平静）
+   - 周围环境与背景物品（电视开着/桌上水杯/地上有玩具等，辅助判断场景）
+4. people: 该时刻出现的人物标识。已知成员用真名，未知人物用「人物A」「人物B」
+   本视频内连续编号（同一人保持同一编号）。
+5. people_mentioned: 必须等于 events 中所有 people 字段出现过的标识去重后的集合。
+   一致性强制：events 里出现的标识必须都在 people_mentioned 里，反之亦然。
+6. is_attention_event: 跌倒、危险动作、异常哭闹、陌生人闯入、身体不适等需关注事件
+   填 true，否则 false。关注事件的 event 仍按上述密度规则抽取，但 description 须明确
+   说明"异常"点（如"张三在 00:01:15 跌坐在地，身体向右侧倾，双手撑地"）。
+7. global_summary: 客观描述，不猜测、不想象、不编造。须包含：谁在画面中、主要活动、
+   是否有关注事件、时段大致结构。
+
+【已知家庭成员】
+按特征匹配，匹配到用真名，匹配不到用「人物X」临时编号：
+{members_block}{camera_hint}{start_hint}"""
+
+
+def build_chat_prompt(context: str, members: str, question: str,
+                      queried_person: str) -> str:
+    """构建智能问答 system prompt（fam-core chat_handler 用）。
+
+    Args:
+      context: sync_events 格式化后的上下文文本（每行一条事件）
+      members: known_members_context 文本
+      question: 用户原始问题
+      queried_person: 用户指定的人物
+    """
+    members_block = members.strip() if members and members.strip() else queried_person
+    context_block = context.strip() if context and context.strip() else "（今日无该人员的监控记录）"
+
+    return f"""你是家庭监控助手。仅根据下方监控数据回答用户问题。
+
+【监控数据】（按时间顺序，每行一条事件）
+{context_block}
+
+【已知家庭成员】
+{members_block}
+
+【用户问题】
+{question}
+（用户关心的人物：{queried_person}）
+
+【回答要求】
+1. 只基于上述监控数据，不编造、不补充数据外信息。
+2. 按时间顺序组织回答，突出关键事件。
+3. 若有关注事件（跌倒、哭闹、陌生人等），重点提示。
+4. 若数据为空或当天未观察到 {queried_person}，明确说"今天没有观察到{queried_person}"。
+5. 用自然语言回答，不要输出 JSON 或列表格式。"""
+
+
+def build_person_merge_prompt(unnamed_lines: str) -> str:
+    """构建人物合并 prompt（person_service._llm_merge 用）。
+
+    Args:
+      unnamed_lines: 待合并人物的场景描述，每行 "- label：出现场景 ..."
+    """
+    return f"""你是家庭监控人物汇总助手。下面是若干人物标识及其出现场景描述。
+请判断哪些标识指向同一个人，并为每个人输出一个稳定的规范名。
+
+【输出格式】
+只输出一个合法 JSON 对象，首字符必须是 {{，禁止 markdown 和解释。
+格式：{{"<原标识>": "<规范名>", ...}}
+
+【命名规则】
+1. 同一人的多个标识合并为同一个规范名。
+2. 规范名用「人物A」「人物B」「人物C」这类占位（按出现频率/首次出现排序），
+   不要编造真实姓名。
+3. 无法判断是否同一人的，保守不合并（各保留独立规范名）。
+4. 规范名必须在输出中唯一：多个原标识可映射到同一规范名，但同一规范名只指向一个人。
+
+【待处理人物】
+{unnamed_lines}"""

fam-edge/src/fam_edge/model_adapters/gemini_adapter.py

@@ -20,6 +20,8 @@ from .base_adapter import BaseModelAdapter
 from .circuit_breaker import CircuitBreaker
 from ..logger import setup_logger
 from ..ai_orchestrator.json_parser import parse_vlm_json, VLMOutputInvalidError
+from ..ai_orchestrator.prompts import build_video_prompt
+from ..config_loader import load_config
 
 logger = setup_logger('fam-edge.gemini_adapter')
 
@@ -317,32 +319,8 @@ class GeminiAdapter(BaseModelAdapter):
         }
 
     def _build_video_prompt(self, known_members: str, event_start_time: str) -> str:
-        start_hint = ""
-        if event_start_time:
-            start_hint = f"\n视频开始时间（北京时间）约为：{event_start_time}。"
-        return f"""你是家庭监控视频分析助手。下面是一段完整监控录像（已整段上传）。
-请观看整段视频，提取其中有用的信息，只输出合法 JSON（不要 markdown、不要任何解释文字），结构如下：
-
-{{
-  "global_summary": "整个时段的整体摘要，简体中文，2-4 句，客观描述人物与主要活动",
-  "events": [
-    {{
-      "timestamp": "事件在视频内的相对时间点(格式 HH:MM:SS，从视频开头 00:00:00 算起)",
-      "description": "该时间点的画面/动作信息摘要（谁、在做什么、位置）",
-      "people": ["出现在该时刻的人物，用已知成员真名或'人物A'/'人物B'"],
-      "is_attention_event": false
-    }}
-  ],
-  "people_mentioned": ["本视频出现过的所有人物标识/真名"]
-}}{start_hint}
-
-规则：
-1. 只描述客观画面，不要猜测或想象。
-2. events 提取视频中"有意义的时间点"（人物出现/动作变化/异常），不要逐秒罗列；timestamp 必须是"视频内相对时间"（如 00:05:23 表示视频开始后 5 分 23 秒），不要输出绝对日期时间。
-3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）：
-{known_members or '（暂无已知成员）'}
-4. is_attention_event：是否为跌倒、危险、异常哭闹等需关注事件（没有则为 false）。
-5. 没有人物出现的时段不要单独成 event；people 留空数组。"""
+        camera = load_config().get('gdrive_sync', {}).get('camera_name', '')
+        return build_video_prompt(known_members, event_start_time, camera)
 
     # ------------------------------------------------------------------
     # 智能问答：纯文本

fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py

@@ -19,6 +19,8 @@ from typing import Dict, List, Optional
 from .base_adapter import BaseModelAdapter
 from .circuit_breaker import CircuitBreaker
 from ..logger import setup_logger
+from ..ai_orchestrator.prompts import build_video_prompt
+from ..config_loader import load_config
 
 logger = setup_logger('fam-edge.nvidia_adapter')
 
@@ -244,31 +246,8 @@ class NvidiaVisionAdapter(BaseModelAdapter):
         return None
 
     def _build_video_prompt(self, known_members: str, event_start_time: str) -> str:
-        start_hint = ""
-        if event_start_time:
-            start_hint = f"\n视频开始时间（北京时间）约为：{event_start_time}。"
-        return f"""你是家庭监控视频分析助手。下面是一段完整监控录像（已整段上传）。
-请观看整段视频，提取其中有用的信息，只输出合法 JSON（不要 markdown、不要解释），结构如下：
-
-{{
-  "global_summary": "整个时段的整体摘要，简体中文，2-4 句",
-  "events": [
-    {{
-      "timestamp": "事件在视频内的相对时间点(格式 HH:MM:SS，从视频开头 00:00:00 算起)",
-      "description": "该时刻画面/动作信息摘要",
-      "people": ["出现在该时刻的人物，用已知成员真名或'人物A'"],
-      "is_attention_event": false
-    }}
-  ],
-  "people_mentioned": ["本视频出现过的所有人物标识/真名"]
-}}{start_hint}
-
-规则：
-1. 只描述客观画面，不猜测。
-2. events 提取有意义的时间点（人物出现/动作变化/异常），timestamp 必须是"视频内相对时间"（如 00:05:23 表示视频开始后 5 分 23 秒），不要输出绝对日期时间。
-3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）：
-{known_members or '（暂无已知成员）'}
-4. is_attention_event：跌倒、危险、异常哭闹等需关注事件（没有则为 false）。"""
+        camera = load_config().get('gdrive_sync', {}).get('camera_name', '')
+        return build_video_prompt(known_members, event_start_time, camera)
 
     # ------------------------------------------------------------------
     # 智能问答：纯文本

fam-edge/src/fam_edge/person_service.py

@@ -20,6 +20,7 @@ from typing import Dict, List, Optional
 from .logger import setup_logger
 from .config_loader import load_config
 from .model_adapters.adapter_factory import build_adapters
+from .ai_orchestrator.prompts import build_person_merge_prompt
 from . import oracle_db
 
 logger = setup_logger('fam-edge.person_service')
@@ -106,14 +107,7 @@ class PersonService:
             label = r['label']
             desc = '；'.join(samples.get(label, [])) or '（无描述）'
             lines.append(f"- {label}：出现场景 {desc}")
-        prompt = f"""你是家庭监控人物汇总助手。下面是若干人物标识及其出现场景描述。
-请判断哪些标识指向同一个人，并为每个人输出一个稳定的规范名（用'人物A'/'人物B'这类占位，
-或若场景描述足以区分则保留原标识）。只输出 JSON，格式：
-{{"<原标识>": "<规范名>", ...}}
-不要编造真实姓名，仅做去重/合并。
-
-待处理人物：
-{chr(10).join(lines)}"""
+        prompt = build_person_merge_prompt(chr(10).join(lines))
         try:
             text = self._llm.chat(prompt, max_tokens=1024)
         except Exception as e: