fix(demo): 收口核查导航、对比分析与问答体验

backend/app/agents/nodes/_answer_headline.py

@@ -336,7 +336,11 @@ def _build_interpretation_segments(state: AgentState, claims: list) -> list[str]
                 f"{len(fields)} 个字段覆盖率仅 {pct}%（有效 {valid}/{total} 期）"
             )
         limitations.extend(other_warnings)
+    plan = state.get("plan")
+    requested_modules = set(getattr(plan, "requested_modules", []) or [])
     for name, ms in (state.get("module_status") or {}).items():
+        if requested_modules and name not in requested_modules:
+            continue
         if getattr(ms, "state", "") in ("partial", "failed"):
             # 8/23 可读性：模块状态转中文（"模块 events 状态: partial"
             # 用户看不懂 → "舆情事件模块部分完成，数据可能不完整"）

backend/app/agents/nodes/build_claims.py

@@ -16,15 +16,33 @@
 _VALID_SEVERITIES = frozenset({"red", "orange", "yellow", "blue", "green", "unknown"})
 
 
-def _collect_evidence(results) -> list[EvidenceRef]:
-    """从模块结果中汇总所有 Evidence。"""
+def _requested_modules(state: AgentState) -> set[str]:
+    """返回本轮明确请求的模块；空集表示非模块型问答或综合兼容路径。"""
+    plan = state.get("plan")
+    return set(getattr(plan, "requested_modules", []) or [])
+
+
+def _collect_evidence(results, requested_modules: set[str]) -> list[EvidenceRef]:
+    """从本轮请求范围内的模块结果汇总 Evidence。"""
     evidence: list[EvidenceRef] = []
     if results:
-        if results.finance and results.finance.evidence:
+        if (
+            results.finance
+            and results.finance.evidence
+            and (not requested_modules or "finance" in requested_modules)
+        ):
             evidence.extend(results.finance.evidence)
-        if results.equity and results.equity.evidence:
+        if (
+            results.equity
+            and results.equity.evidence
+            and (not requested_modules or "equity" in requested_modules)
+        ):
             evidence.extend(results.equity.evidence)
-        if results.events and results.events.evidence:
+        if (
+            results.events
+            and results.events.evidence
+            and (not requested_modules or "events" in requested_modules)
+        ):
             evidence.extend(results.events.evidence)
     return evidence
 
@@ -388,16 +406,22 @@ def build_claims_node(state: AgentState) -> dict:
     company = state.get("company")
     company_name = company.sec_name if company else "目标公司"
     claims: list[Claim] = []
+    requested_modules = _requested_modules(state)
 
     # 收集 evidence 并建立索引（按 evidence_id 去重，保留顺序）
-    evidence = _collect_evidence(results)
+    evidence = _collect_evidence(results, requested_modules)
     evidence_index: dict[str, EvidenceRef] = {}
     for ev in evidence:
         if ev.evidence_id not in evidence_index:
             evidence_index[ev.evidence_id] = ev
 
     # ── 财务 Claim ───────────────────────────────────────
-    if results and results.finance and results.finance.rule_statuses:
+    if (
+        results
+        and results.finance
+        and results.finance.rule_statuses
+        and (not requested_modules or "finance" in requested_modules)
+    ):
         rule_details = results.finance.rule_details or {}
         for ordinal, (rule_id, status) in enumerate(
             results.finance.rule_statuses.items()
@@ -438,16 +462,21 @@ def build_claims_node(state: AgentState) -> dict:
         results
         and results.equity
         and (results.equity.chains or results.equity.chain_details)
+        and (not requested_modules or "equity" in requested_modules)
     ):
         _append_equity_claims(state, claims, evidence_index, company_name)
 
     # ── 事件 Claim（公告、评级、事件簇各自只绑定可回查证据） ──
-    if results and results.events:
+    if (
+        results
+        and results.events
+        and (not requested_modules or "events" in requested_modules)
+    ):
         _append_event_claims(state, claims, evidence_index, company_name)
 
     # ── 交叉验证 Claim（读取 state.cross_validation，B4/B3 联动） ──
     cross_validation = state.get("cross_validation")
-    if cross_validation is not None:
+    if cross_validation is not None and len(requested_modules) != 1:
         checks = cross_validation.checks or []
         for ordinal, check in enumerate(checks):
             if check.status != "fail":

backend/app/agents/nodes/finance.py

@@ -341,6 +341,7 @@ def finance_node(state: AgentState) -> dict:
             "explanation": str(r.explanation or ""),
             "severity": r.severity or "",
             "current": dict(getattr(r, "current", None) or {}),
+            "history": list(getattr(r, "history", None) or []),
             "evidence_ids": [],
             "quality": dict(r.quality or {}),
             "calculation_trace": (

backend/app/agents/nodes/generate_answer.py

@@ -190,10 +190,57 @@
 logger = logging.getLogger(__name__)
 
 
+_MODULE_CLAIM_TYPES = {
+    "finance": {"financial"},
+    "equity": {"equity"},
+    "events": {"event"},
+}
+_EVIDENCE_MODULE_BY_SOURCE_TYPE = {
+    "financial_statement": "finance",
+    "neo4j_relationship": "equity",
+    "ownership_record": "equity",
+    "announcement": "events",
+    "event_cluster": "events",
+    "news_article": "events",
+    "regulation": "events",
+    "web_search": "events",
+}
+
+
+def _scope_analysis_items(
+    state: AgentState, claims: list, evidence: list
+) -> tuple[list, list, set[str]]:
+    """按本轮请求模块过滤展示材料，阻断异常上游结果串入单模块回答。"""
+    plan = state.get("plan")
+    requested_modules = set(getattr(plan, "requested_modules", []) or [])
+    if not requested_modules:
+        return claims, evidence, requested_modules
+
+    allowed_claim_types = set().union(
+        *(_MODULE_CLAIM_TYPES.get(module, set()) for module in requested_modules)
+    )
+    if len(requested_modules) > 1:
+        allowed_claim_types.add("cross_validation")
+    scoped_claims = [
+        claim
+        for claim in claims
+        if getattr(claim, "claim_type", "") in allowed_claim_types
+    ]
+    scoped_evidence = []
+    for item in evidence:
+        module = getattr(item, "module", "") or _EVIDENCE_MODULE_BY_SOURCE_TYPE.get(
+            getattr(item, "source_type", ""), ""
+        )
+        if module in requested_modules:
+            scoped_evidence.append(item)
+    return scoped_claims, scoped_evidence, requested_modules
+
+
 def generate_answer_node(state: AgentState) -> dict:
     company = state.get("company")
-    claims = state.get("claims", [])
-    evidence = state.get("evidence", [])
+    claims, evidence, requested_modules = _scope_analysis_items(
+        state, state.get("claims", []), state.get("evidence", [])
+    )
     finance_ran, finance = _finance_executed(state)
     finance_blocked = _finance_all_blocked(finance)
     finance_unknown_type = finance_blocked and any(
@@ -382,9 +429,9 @@ def generate_answer_node(state: AgentState) -> dict:
                 chitchat_answer = (
                     f"{intro}我可以核查财务勾稽、股权控制链和公告舆情。\n"
                     "你可以这样问：\n"
-                    "- 分析康美药业 2025 年报的财务异常\n"
-                    "- 查看金牌家居的实际控制人链路\n"
-                    "- 核对贵州茅台近期公告与财务数据\n"
+                    "- 分析金牌家居财务风险\n"
+                    "- 查看比亚迪经营现金流\n"
+                    "- 查看金牌家居股权穿透\n"
                     "请输入上市公司名称或股票代码开始分析。"
                 )
             _emit_segment(state, chitchat_answer)
@@ -406,7 +453,7 @@ def generate_answer_node(state: AgentState) -> dict:
             elif any(kw in ql for kw in _CAPABILITY_KW):
                 guide_answer = (
                     "我可以核查上市公司的财务勾稽、股权控制链和公告舆情。"
-                    "例如可以问“分析康美药业 2025 年报”或“查看金牌家居的实控人链路”。"
+                    "例如可以问“分析金牌家居财务风险”或“查看金牌家居股权穿透”。"
                     "请输入上市公司名称或股票代码开始分析。"
                 )
             else:
@@ -686,6 +733,10 @@ def generate_answer_node(state: AgentState) -> dict:
                 )
             )
         )
+    elif mode == "finance" and risk_count:
+        conclusion = (
+            name_code + "财务分析完成，" + RISK_SIGNAL_IN_SCOPE.format(n=risk_count)
+        )
     elif risk_count:
         if finance_ran:
             # 口径限定：本分析基于母公司报表及当前数据覆盖
@@ -718,7 +769,11 @@ def generate_answer_node(state: AgentState) -> dict:
     # ② 多类核心信号摘要（含评级/交叉验证/综合风险）
     risk_output = state.get("risk_output")
     results = state.get("results")
-    summary = _build_signal_summary(claims, results=results, risk_output=risk_output)
+    # 单模块问答不得借用综合风险评分（可能包含未请求模块的信号）。
+    scoped_risk_output = risk_output if len(requested_modules) != 1 else None
+    summary = _build_signal_summary(
+        claims, results=results, risk_output=scoped_risk_output
+    )
     # ③ 财务触发规则明细（V12 §4.3 规则触发清单）
     rule_details = _build_rule_details(state)
 
@@ -773,7 +828,7 @@ def append_segment(text: str) -> None:
     if summary:
         seg = summary + "。"
         append_segment(seg)
-    brief = _build_company_brief_analysis(state, claims, risk_output=risk_output)
+    brief = _build_company_brief_analysis(state, claims, risk_output=scoped_risk_output)
     if brief:
         append_segment(brief)
     append_segment(_build_cross_module_observation(state, claims))
@@ -872,11 +927,16 @@ def append_segment(text: str) -> None:
         answer = _polish_answer(answer)
 
     # 风险等级：优先使用 risk 节点输出（否则回退 claim 最高严重度）
-    risk_level = (
-        (getattr(risk_output, "risk_level", "") or _highest_severity(claims))
-        if (risk_output is not None or claims)
-        else "unknown"
-    )
+    if len(requested_modules) == 1:
+        # 单模块不复用综合评分；但该模块全部未执行时也不能把空 Claim
+        # 的默认 green 误当作“正常”。
+        risk_level = "unknown" if finance_blocked else _highest_severity(claims)
+    else:
+        risk_level = (
+            getattr(risk_output, "risk_level", "") or _highest_severity(claims)
+            if (risk_output is not None or claims)
+            else "unknown"
+        )
 
     # #4：研报 Claim/Evidence 合并进 FinalResponse（reducer 会再合并进
     # AgentState，validate_evidence → persist_turn 因此可完整落库/回查）

backend/app/agents/nodes/persist_turn.py

@@ -506,7 +506,8 @@ def persist_turn_node(state: AgentState) -> dict:
     module_status_json = _to_json(state.get("module_status", {}))
     panel_data_json = _to_json(_build_panel_data(state))
     response_meta_json = _to_json(_build_response_meta(state))
-    title = question[:30]
+    # 会话列表展示两行主题；保留更多首问上下文，避免相近的分析问题难以区分。
+    title = question[:48]
     provenance_ok = True
     provenance_error = ""
 

backend/app/api/v1/routers/chat.py

@@ -409,7 +409,12 @@ def _build_chat_response(
     从 Agent State 中提取结构化数据并转换为 API DTO。
     """
     final_response = result.get("final_response")
-    evidence = result.get("evidence", [])
+    # 回答层已按本轮请求模块裁剪展示材料；REST 不得重新透出未选模块证据。
+    evidence = (
+        getattr(final_response, "evidence", [])
+        if final_response is not None
+        else result.get("evidence", [])
+    )
     module_status = result.get("module_status", {})
     results = result.get("results")
 
@@ -483,7 +488,12 @@ def _build_chat_response(
         follow_ups = getattr(final_response, "follow_ups", []) or []
 
     # claims — 从 Agent State 透出（结构化问答结论声明，API 公共投影）
-    claims_items = [ClaimV1.from_claim(c) for c in result.get("claims", [])]
+    response_claims = (
+        getattr(final_response, "claims", [])
+        if final_response is not None
+        else result.get("claims", [])
+    )
+    claims_items = [ClaimV1.from_claim(c) for c in response_claims]
     # module_status — typed ModuleStatusV1（对象/dict/字符串/None 全兼容）
     module_status_items = {
         k: ModuleStatusV1.from_status(v) for k, v in module_status.items()

backend/app/api/v1/routers/provenance.py

@@ -93,9 +93,9 @@ def _claims_for_evidence(evidence_id: str) -> list[dict]:
         rows = (
             conn.execute(
                 text(
-                    "SELECT c.* FROM claims c "
+                    "SELECT DISTINCT c.* FROM claims c "
                     "JOIN claim_evidence_links l ON l.claim_id = c.claim_id "
-                    "WHERE l.evidence_id = :eid ORDER BY l.sequence_no, c.claim_id"
+                    "WHERE l.evidence_id = :eid ORDER BY c.claim_id"
                 ),
                 {"eid": evidence_id},
             )

backend/app/api/v1/routers/risk.py

@@ -27,6 +27,7 @@
     FraudConclusionData,
     ImpactAdviceData,
     ImpactAdviceSegmentData,
+    VerificationNavigationItemData,
     MitigatingFactor,
     PatternMatch,
     RiskEvidence,
@@ -463,6 +464,18 @@ async def get_impact_advice(
                 )
                 for s in result.segments
             ],
+            verification_navigation=[
+                VerificationNavigationItemData(
+                    rule_id=item.rule_id,
+                    rule_name=item.rule_name,
+                    severity=item.severity,
+                    explanation=item.explanation,
+                    quantified_context=item.quantified_context,
+                    actions=item.actions,
+                    evidence_ids=item.evidence_ids,
+                )
+                for item in result.verification_navigation
+            ],
             evidence_count=result.evidence_count,
             warnings=result.warnings,
         ),

backend/app/api/v1/schemas/risk.py

@@ -94,6 +94,7 @@ class DerivationSignal(BaseModel):
     severity: str = Field(default="unknown")
     explanation: str = Field(default="")
     current: dict = Field(default_factory=dict)
+    history: list[dict] = Field(default_factory=list)
     industry_percentile: float | None = Field(default=None)
     data_refs: list[DerivationDataRef] = Field(default_factory=list)
     evidence_ids: list[str] = Field(default_factory=list)
@@ -168,6 +169,18 @@ class ImpactAdviceSegmentData(BaseModel):
     evidence_ids: list[str] = Field(default_factory=list, description="可回查证据 ID")
 
 
+class VerificationNavigationItemData(BaseModel):
+    """画像页核查导航：L0 风险点 + L1 量化参考 + L2 核查动作。"""
+
+    rule_id: str = Field(default="")
+    rule_name: str = Field(default="")
+    severity: str = Field(default="unknown")
+    explanation: str = Field(default="")
+    quantified_context: str = Field(default="")
+    actions: list[str] = Field(default_factory=list)
+    evidence_ids: list[str] = Field(default_factory=list)
+
+
 class ImpactAdviceData(BaseModel):
     """Phase E 会3：影响与建议聚合（画像页影响建议模块数据源）。"""
 
@@ -179,5 +192,8 @@ class ImpactAdviceData(BaseModel):
     overall_advice: str = Field(default="", description="整体建议（LLM 或模板）")
     method: str = Field(default="template", description="llm | template")
     segments: list[ImpactAdviceSegmentData] = Field(default_factory=list)
+    verification_navigation: list[VerificationNavigationItemData] = Field(
+        default_factory=list
+    )
     evidence_count: int = Field(default=0)
     warnings: list[str] = Field(default_factory=list)