1418 lines
50 KiB
Python
1418 lines
50 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import timedelta
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from sqlalchemy import JSON
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
from app.models.base import Base
|
||
from app.models.feishu_reply_outbox import FeishuReplyOutbox
|
||
from app.models.ingest_event import IngestEvent
|
||
from app.models.qa_record import QARecord
|
||
from app.models.qa_session import QASession
|
||
from app.models.qa_session_state import QASessionState
|
||
from app.services.adaptive_rag_service.mixin_core_intent import AdaptiveRAGCoreIntentMixin
|
||
from app.services.adaptive_rag_service.mixin_followup_anchor import (
|
||
AdaptiveRAGFollowupAnchorMixin,
|
||
FollowupAnchorResolution,
|
||
)
|
||
from app.services.adaptive_rag_service.mixin_qa_session_core import AdaptiveRAGQASessionCoreMixin
|
||
from app.services.adaptive_rag_service.mixin_session_state import AdaptiveRAGSessionStateMixin
|
||
from app.utils.timezone import now_cn_naive
|
||
|
||
|
||
def _use_sqlite_json_types() -> None:
|
||
for table in (QARecord.__table__, QASessionState.__table__, IngestEvent.__table__):
|
||
for column in table.columns:
|
||
if column.name in {
|
||
"sources",
|
||
"domains_used",
|
||
"citations",
|
||
"sources_used",
|
||
"evidence_pack",
|
||
"tool_trace",
|
||
"retrieval_metrics",
|
||
"last_entity_ids",
|
||
"rewrite_history",
|
||
"active_workspace",
|
||
"context_capsules",
|
||
"pending_clarification",
|
||
"routing_snapshot",
|
||
}:
|
||
column.type = JSON()
|
||
if table.name == "ingest_events" and column.name == "payload":
|
||
column.type = JSON()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def db_session():
|
||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
_use_sqlite_json_types()
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(
|
||
lambda sync_conn: Base.metadata.create_all(
|
||
sync_conn,
|
||
tables=[
|
||
QASession.__table__,
|
||
QASessionState.__table__,
|
||
QARecord.__table__,
|
||
FeishuReplyOutbox.__table__,
|
||
IngestEvent.__table__,
|
||
],
|
||
)
|
||
)
|
||
maker = async_sessionmaker(engine, expire_on_commit=False)
|
||
async with maker() as session:
|
||
yield session
|
||
await engine.dispose()
|
||
|
||
|
||
class _FakeContextCache:
|
||
async def get_session_messages(self, *, tenant_id: int, session_id: int | None):
|
||
_ = (tenant_id, session_id)
|
||
return []
|
||
|
||
async def append_session_messages(self, *, tenant_id: int, session_id: int | None, question: str, answer: str):
|
||
_ = (tenant_id, session_id, question, answer)
|
||
return None
|
||
|
||
@staticmethod
|
||
def messages_to_turns(messages):
|
||
_ = messages
|
||
return []
|
||
|
||
|
||
class _Harness(
|
||
AdaptiveRAGCoreIntentMixin,
|
||
AdaptiveRAGQASessionCoreMixin,
|
||
AdaptiveRAGSessionStateMixin,
|
||
AdaptiveRAGFollowupAnchorMixin,
|
||
):
|
||
def __init__(self, db) -> None:
|
||
super().__init__(db, tenant_id=1)
|
||
self.context_cache = _FakeContextCache()
|
||
|
||
async def _resolve_followup_anchor_semantics(
|
||
self,
|
||
*,
|
||
message: str,
|
||
has_explicit_reply_anchor: bool,
|
||
recent_candidates: list[dict[str, Any]],
|
||
) -> dict[str, Any]:
|
||
text = str(message or "").strip()
|
||
normalized = text.replace(" ", "")
|
||
if "项目有吗" in normalized or "有没有" in normalized:
|
||
return {
|
||
"followup_mode": "none",
|
||
"meta_followup_kind": "none",
|
||
"anchor_strategy": "none",
|
||
"candidate_index": 0,
|
||
"confidence": 0.9,
|
||
}
|
||
|
||
meta_kind = "none"
|
||
if any(token in normalized for token in ("哪个模型", "什么模型", "谁回复", "谁回答")):
|
||
meta_kind = "reply_model_identity"
|
||
elif any(token in normalized for token in ("什么情况", "怎么回事", "为什么失败", "为什么报错", "详细说", "详细解释")):
|
||
meta_kind = "reply_error_reason"
|
||
elif any(token in normalized for token in ("来源", "根据什么", "什么依据")):
|
||
meta_kind = "reply_source"
|
||
|
||
if meta_kind != "none":
|
||
followup_mode = "meta"
|
||
elif has_explicit_reply_anchor or recent_candidates:
|
||
followup_mode = "referential"
|
||
else:
|
||
followup_mode = "none"
|
||
|
||
if has_explicit_reply_anchor:
|
||
anchor_strategy = "explicit"
|
||
elif meta_kind != "none" and len(recent_candidates) > 1:
|
||
anchor_strategy = "clarify"
|
||
elif len(recent_candidates) == 1:
|
||
anchor_strategy = "implicit_recent"
|
||
elif len(recent_candidates) > 1:
|
||
anchor_strategy = "clarify"
|
||
else:
|
||
anchor_strategy = "none"
|
||
|
||
return {
|
||
"followup_mode": followup_mode,
|
||
"meta_followup_kind": meta_kind,
|
||
"anchor_strategy": anchor_strategy,
|
||
"candidate_index": 1 if anchor_strategy == "implicit_recent" and recent_candidates else 0,
|
||
"confidence": 0.9,
|
||
}
|
||
|
||
async def _resolve_pending_clarification_selection_semantics(
|
||
self,
|
||
*,
|
||
question: str,
|
||
payload: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
text = str(question or "").strip()
|
||
normalized = text.replace(" ", "")
|
||
kind = str(payload.get("kind") or "").strip().lower()
|
||
if kind == "query_route_selection":
|
||
selected_key = ""
|
||
if "外部" in normalized or normalized.upper() == "A":
|
||
selected_key = "A"
|
||
elif "内部" in normalized or normalized.upper() == "B":
|
||
selected_key = "B"
|
||
return {
|
||
"selection_status": "resolved" if selected_key else "unresolved",
|
||
"selected_key": selected_key,
|
||
"selected_index": 0,
|
||
"confidence": 0.9 if selected_key else 0.0,
|
||
"reason": "test_double",
|
||
}
|
||
|
||
selected_index = 0
|
||
if any(token in normalized for token in ("第二", "第2", "2")):
|
||
selected_index = 2
|
||
elif any(token in normalized for token in ("第一", "第1", "1")):
|
||
selected_index = 1
|
||
return {
|
||
"selection_status": "resolved" if selected_index else "unresolved",
|
||
"selected_key": "",
|
||
"selected_index": selected_index,
|
||
"confidence": 0.9 if selected_index else 0.0,
|
||
"reason": "test_double",
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_external_conversation_key_creates_and_reuses_real_qa_session(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
|
||
session_id, turn_index = await service._touch_session_for_new_record(
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
question="帮我解释下系统架构",
|
||
external_conversation_key="feishu:chat-1:root-1",
|
||
)
|
||
|
||
assert session_id is not None
|
||
assert turn_index == 1
|
||
|
||
reused_session_id, reused_turn_index = await service._touch_session_for_new_record(
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
question="系统有什么模型？",
|
||
external_conversation_key="feishu:chat-1:root-1",
|
||
)
|
||
|
||
assert reused_session_id == session_id
|
||
assert reused_turn_index == 2
|
||
|
||
row = await db_session.get(QASession, int(session_id))
|
||
assert row is not None
|
||
assert row.user_id is None
|
||
assert row.external_conversation_key == "feishu:chat-1:root-1"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_meta_followup_prefers_single_recent_reply_candidate(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="系统架构",
|
||
external_conversation_key="feishu:chat-1:root-1",
|
||
started_at=now_cn_naive(),
|
||
last_message_at=now_cn_naive(),
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add(
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="帮我解释下系统的架构",
|
||
answer="PMFlow 系统采用前后端分离架构。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=10,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="我问你这个是哪个模型执行回复的？",
|
||
session_id=int(session.id),
|
||
context_session_id=int(session.id),
|
||
session_state=state,
|
||
)
|
||
|
||
assert resolution.anchor_mode == "implicit_recent_reply"
|
||
assert resolution.anchor_reply_id is not None
|
||
assert resolution.is_meta_followup is True
|
||
assert resolution.candidate_replies == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_followup_anchor_prefers_structured_semantics_without_keyword_hints(
|
||
db_session,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="架构说明",
|
||
external_conversation_key="feishu:chat-1:root-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add(
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="帮我解释下系统架构",
|
||
answer="PMFlow 系统采用前后端分离架构。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=10,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
conversation_scope_key="feishu:chat-1:root-1",
|
||
created_at=now,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
async def _fake_semantics(**kwargs):
|
||
_ = kwargs
|
||
return {
|
||
"followup_mode": "meta",
|
||
"meta_followup_kind": "reply_explanation",
|
||
"anchor_strategy": "implicit_recent",
|
||
"candidate_index": 1,
|
||
"confidence": 0.93,
|
||
}
|
||
|
||
monkeypatch.setattr(
|
||
service,
|
||
"_resolve_followup_anchor_semantics",
|
||
_fake_semantics,
|
||
raising=False,
|
||
)
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="沿着前一条把细节铺开",
|
||
session_id=int(session.id),
|
||
context_session_id=int(session.id),
|
||
session_state=state,
|
||
conversation_scope_key="feishu:chat-1:root-1",
|
||
)
|
||
|
||
assert resolution.anchor_mode == "implicit_recent_reply"
|
||
assert resolution.is_meta_followup is True
|
||
assert resolution.meta_followup_kind == "reply_explanation"
|
||
assert resolution.anchor_reply is not None
|
||
assert resolution.anchor_reply["question"] == "帮我解释下系统架构"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ambiguous_recent_reply_returns_numbered_candidates(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="新会话",
|
||
external_conversation_key="feishu:chat-1:root-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=2,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add_all(
|
||
[
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="帮我解释下系统架构",
|
||
answer="PMFlow 系统采用前后端分离架构。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=10,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
created_at=now - timedelta(minutes=1),
|
||
),
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="系统有什么模型？",
|
||
answer="当前有 low/high 两层模型路由。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=12,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=2,
|
||
created_at=now,
|
||
),
|
||
]
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="那你刚才那条是哪个模型回复的？",
|
||
session_id=int(session.id),
|
||
context_session_id=int(session.id),
|
||
session_state=state,
|
||
)
|
||
|
||
assert resolution.anchor_mode == "ambiguous_candidates"
|
||
assert resolution.anchor_reply_id is None
|
||
assert len(resolution.candidate_replies) == 2
|
||
assert resolution.candidate_replies[0]["index"] == 1
|
||
assert resolution.candidate_replies[1]["index"] == 2
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_pending_clarification_selection_restores_original_question(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="新会话",
|
||
external_conversation_key="feishu:chat-1:root-1",
|
||
started_at=now_cn_naive(),
|
||
last_message_at=now_cn_naive(),
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
state.pending_clarification = {
|
||
"kind": "followup_anchor_selection",
|
||
"original_question": "我问你这个是哪个模型执行回复的？",
|
||
"candidates": [
|
||
{"index": 1, "qa_id": 101, "label": "系统架构那条"},
|
||
{"index": 2, "qa_id": 102, "label": "系统模型那条"},
|
||
],
|
||
}
|
||
await db_session.commit()
|
||
|
||
resolved = await service._consume_pending_clarification_selection(
|
||
question="第二条",
|
||
session_state=state,
|
||
)
|
||
|
||
assert resolved is not None
|
||
assert resolved["selected_qa_id"] == 102
|
||
assert resolved["effective_question"] == "我问你这个是哪个模型执行回复的？"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_selected_followup_clarification_anchor_binds_exact_qa_record(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="新会话",
|
||
external_conversation_key="feishu:chat-1:root-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=2,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add_all(
|
||
[
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="帮我解释下系统架构",
|
||
answer="PMFlow 系统采用前后端分离架构。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=10,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
created_at=now - timedelta(minutes=1),
|
||
),
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="系统有什么模型？",
|
||
answer="当前有 low/high 两层模型路由。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=12,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=2,
|
||
created_at=now,
|
||
),
|
||
]
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="帮我继续说",
|
||
session_id=int(session.id),
|
||
context_session_id=int(session.id),
|
||
session_state=state,
|
||
selected_anchor_reply_id=2,
|
||
)
|
||
|
||
assert resolution.anchor_mode == "clarification_selection"
|
||
assert resolution.anchor_reply_id == 2
|
||
assert resolution.anchor_reply is not None
|
||
assert resolution.anchor_reply["question"] == "系统有什么模型？"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_anchor_clarification_prefers_reply_or_object_id_guidance(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
|
||
payload = service._build_followup_clarification_payload(
|
||
question="沿着刚才那个继续说",
|
||
anchor_resolution=FollowupAnchorResolution(clarification_code="no_anchor_found"),
|
||
session_state=None,
|
||
)
|
||
|
||
assert payload is not None
|
||
assert "引用那条消息" in payload["answer"]
|
||
assert "对象 ID" in payload["answer"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_explicit_reply_to_message_id_resolves_anchor_without_session_history(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
db_session.add(
|
||
IngestEvent(
|
||
tenant_id=1,
|
||
channel="feishu_message",
|
||
event_key="om_user_message_1",
|
||
status="done",
|
||
payload={"message_text": "帮我解释下系统的架构"},
|
||
)
|
||
)
|
||
db_session.add(
|
||
FeishuReplyOutbox(
|
||
tenant_id=1,
|
||
trace_id="trace-1",
|
||
source_message_id="om_user_message_1",
|
||
sent_message_id="om_bot_reply_1",
|
||
chat_id="oc-chat-1",
|
||
message_text='<at user_id="ou-1">林宇航</at> PMFlow 系统采用前后端分离架构。',
|
||
status="sent",
|
||
created_at=now - timedelta(minutes=1),
|
||
sent_at=now - timedelta(minutes=1),
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="这个是哪个模型去回复我问题的？",
|
||
session_id=None,
|
||
context_session_id=None,
|
||
session_state=None,
|
||
thread_context={"reply_to_message_id": "om_bot_reply_1"},
|
||
)
|
||
|
||
assert resolution.anchor_mode == "explicit"
|
||
assert resolution.is_meta_followup is True
|
||
assert resolution.anchor_reply is not None
|
||
assert resolution.anchor_reply["question"] == "帮我解释下系统的架构"
|
||
assert "前后端分离架构" in resolution.anchor_reply["anchor_reply_preview"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_explicit_reply_anchor_rewrite_takes_priority_over_session_state_subject(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
session_state = QASessionState(
|
||
tenant_id=1,
|
||
session_id=999,
|
||
last_subject="_sync_pmflow_rag_async->backend/app/api/sstc.py",
|
||
last_entity_ids=[],
|
||
last_intent="question",
|
||
ambiguity="clear",
|
||
rewrite_history=[],
|
||
pending_clarification=None,
|
||
pending_clarification_expires_at=None,
|
||
)
|
||
|
||
rewritten, rewrite_meta = service._rewrite_question_with_state(
|
||
question="B",
|
||
conversation=[],
|
||
session_state=session_state,
|
||
anchor_resolution=FollowupAnchorResolution(
|
||
followup_mode="referential",
|
||
anchor_mode="explicit",
|
||
anchor_kind="feishu_reply_outbox",
|
||
anchor_reply={
|
||
"qa_id": 321,
|
||
"question": "你提到的是给运营公司打造「主播推流」的新需求",
|
||
"label": "主播推流新需求",
|
||
"anchor_reply_preview": "这是个典型的宽泛 IT 项目概念，我先给你拆成 A/B/C 三类。",
|
||
"source_message_id": "om_user_message_1",
|
||
},
|
||
),
|
||
)
|
||
|
||
assert rewritten == "围绕上一轮对话“主播推流新需求”继续回答：B"
|
||
assert rewrite_meta["mode"] == "followup_anchor_recent_reply"
|
||
assert rewrite_meta["qa_id"] == 321
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_explicit_reply_falls_back_to_current_ingest_closure_turns_when_outbox_mapping_missing(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
db_session.add(
|
||
IngestEvent(
|
||
tenant_id=1,
|
||
channel="feishu_message",
|
||
event_key="om_current_message_1",
|
||
status="done",
|
||
payload={
|
||
"message_text": "这个是哪个模型去回复我问题的？",
|
||
"closure_turns": [
|
||
{"role": "user", "text": "帮我解释下系统的架构"},
|
||
{"role": "assistant", "text": "PMFlow 系统采用前后端分离架构。"},
|
||
{"role": "user", "text": "这个是哪个模型去回复我问题的？"},
|
||
],
|
||
},
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="这个是哪个模型去回复我问题的？",
|
||
session_id=None,
|
||
context_session_id=None,
|
||
session_state=None,
|
||
source_message_id="om_current_message_1",
|
||
thread_context={"reply_to_message_id": "om_unmapped_reply"},
|
||
)
|
||
|
||
assert resolution.anchor_mode == "explicit"
|
||
assert resolution.is_meta_followup is True
|
||
assert resolution.anchor_reply is not None
|
||
assert resolution.anchor_reply["question"] == "帮我解释下系统的架构"
|
||
assert "前后端分离架构" in resolution.anchor_reply["anchor_reply_preview"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_project_followup_state_prefers_answer_project_name_over_raw_question(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
state = SimpleNamespace(
|
||
last_subject=None,
|
||
last_entity_ids=[],
|
||
last_intent="question",
|
||
ambiguity="clear",
|
||
rewrite_history=[],
|
||
pending_clarification=None,
|
||
pending_clarification_expires_at=None,
|
||
)
|
||
|
||
await service._update_session_state_after_turn(
|
||
session_state=state,
|
||
question="帮我看下采购经理那个需求是否存在系统？",
|
||
effective_question="帮我看下采购经理那个需求是否存在系统？",
|
||
answer=(
|
||
"有，命中1个项目：项目ID 123，项目名“采购经理每日实时库存上报服务”，"
|
||
"状态 active，负责人 系统管理员。"
|
||
),
|
||
query_type="existence",
|
||
ambiguity="clear",
|
||
rewrite_meta={"mode": "direct"},
|
||
sources=[
|
||
{
|
||
"source": "db",
|
||
"label": "Project database",
|
||
"score": 0.99,
|
||
"metadata": {"table": "projects", "matched_project_ids": [123]},
|
||
}
|
||
],
|
||
)
|
||
|
||
assert state.last_subject == "采购经理每日实时库存上报服务"
|
||
assert state.last_entity_ids == [123]
|
||
|
||
rewritten, rewrite_meta = service._rewrite_question_with_state(
|
||
question="这个项目怎么样？",
|
||
conversation=[],
|
||
session_state=state,
|
||
anchor_resolution=FollowupAnchorResolution(
|
||
followup_mode="referential",
|
||
anchor_mode="implicit_recent",
|
||
),
|
||
)
|
||
|
||
assert rewritten == "围绕“采购经理每日实时库存上报服务”回答：这个项目怎么样？"
|
||
assert rewrite_meta["mode"] == "state_subject"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_project_followup_state_extracts_unquoted_project_name_from_answer(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
state = SimpleNamespace(
|
||
last_subject=None,
|
||
last_entity_ids=[],
|
||
last_intent="question",
|
||
ambiguity="clear",
|
||
rewrite_history=[],
|
||
pending_clarification=None,
|
||
pending_clarification_expires_at=None,
|
||
)
|
||
|
||
await service._update_session_state_after_turn(
|
||
session_state=state,
|
||
question="帮我看下采购经理那个需求是否存在系统？",
|
||
effective_question="帮我看下采购经理那个需求是否存在系统？",
|
||
answer=(
|
||
"有，命中 1 个项目：项目ID 123、项目名 采购经理每日实时库存上报服务、"
|
||
"状态 active、负责人 系统管理员。"
|
||
),
|
||
query_type="existence",
|
||
ambiguity="clear",
|
||
rewrite_meta={"mode": "direct"},
|
||
sources=[
|
||
{
|
||
"source": "db",
|
||
"label": "Project database",
|
||
"score": 0.99,
|
||
"metadata": {"table": "projects", "matched_project_ids": [123]},
|
||
}
|
||
],
|
||
)
|
||
|
||
assert state.last_subject == "采购经理每日实时库存上报服务"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_project_followup_state_prefers_customer_visible_project_focus_over_kg_entities(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
state = SimpleNamespace(
|
||
last_subject=None,
|
||
last_entity_ids=[],
|
||
last_intent="question",
|
||
ambiguity="clear",
|
||
rewrite_history=[],
|
||
pending_clarification=None,
|
||
pending_clarification_expires_at=None,
|
||
)
|
||
|
||
await service._update_session_state_after_turn(
|
||
session_state=state,
|
||
question="现在有什么项目",
|
||
effective_question="现在有什么项目",
|
||
answer="你当前有 1 个项目：童析-学龄前儿童行为特质AI评估助手。",
|
||
query_type="project_inventory",
|
||
ambiguity="clear",
|
||
rewrite_meta={"mode": "direct"},
|
||
sources=[
|
||
{
|
||
"source": "graph",
|
||
"label": "Knowledge Graph",
|
||
"score": 0.8,
|
||
"metadata": {
|
||
"subject_entity_id": 14308,
|
||
"object_entity_id": 14311,
|
||
"subject_name": "个人空间",
|
||
"object_name": "/pmflow-os/projects/{project_id}/truth-state",
|
||
},
|
||
},
|
||
{
|
||
"source": "pmflow",
|
||
"label": "项目数据",
|
||
"score": 1.0,
|
||
"metadata": {
|
||
"object_ref": "project:141",
|
||
"project_id": 141,
|
||
"source_type": "customer_visible_object",
|
||
"subject_name": "童析-学龄前儿童行为特质AI评估助手",
|
||
"query_keyword": "童析-学龄前儿童行为特质AI评估助手",
|
||
},
|
||
},
|
||
],
|
||
)
|
||
|
||
assert state.last_subject == "童析-学龄前儿童行为特质AI评估助手"
|
||
assert state.last_entity_ids == [141]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_external_research_workspace_persists_candidate_set_and_survives_followup_rewrite(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
state = SimpleNamespace(
|
||
last_subject=None,
|
||
last_entity_ids=[],
|
||
last_intent="question",
|
||
ambiguity="clear",
|
||
rewrite_history=[],
|
||
active_workspace=None,
|
||
context_capsules=[],
|
||
pending_clarification=None,
|
||
pending_clarification_expires_at=None,
|
||
)
|
||
|
||
await service._update_session_state_after_turn(
|
||
session_state=state,
|
||
question="这个项目的竞品有哪些",
|
||
effective_question="这个项目的竞品有哪些",
|
||
answer="我先给你一组候选竞品。",
|
||
query_type="compare",
|
||
ambiguity="clear",
|
||
rewrite_meta={"mode": "direct"},
|
||
sources=[
|
||
{
|
||
"source": "pmflow",
|
||
"label": "项目数据",
|
||
"score": 1.0,
|
||
"metadata": {
|
||
"object_ref": "project:141",
|
||
"project_id": 141,
|
||
"source_type": "customer_visible_object",
|
||
"subject_name": "童析-学龄前儿童行为特质AI评估助手",
|
||
"query_keyword": "童析-学龄前儿童行为特质AI评估助手",
|
||
},
|
||
}
|
||
],
|
||
route_decision={"effective_source_policy": "external"},
|
||
response_payload={
|
||
"research_workspace": {
|
||
"workspace_id": "ws_competitor_141",
|
||
"workspace_kind": "hybrid_research",
|
||
"topic": "竞品研究",
|
||
"primary_object_ref": "project:141",
|
||
"primary_object_id": 141,
|
||
"primary_object_title": "童析-学龄前儿童行为特质AI评估助手",
|
||
"candidate_set_binding": "cset_competitors_141",
|
||
"candidate_items": [
|
||
{"candidate_ref": "candidate:goally", "display_name": "Goally"},
|
||
{"candidate_ref": "candidate:abcmouse", "display_name": "ABCmouse"},
|
||
],
|
||
}
|
||
},
|
||
)
|
||
|
||
rewritten, meta = service._rewrite_question_with_state(
|
||
question="详细对比下",
|
||
conversation=[],
|
||
session_state=state,
|
||
anchor_resolution=SimpleNamespace(followup_mode="referential", anchor_mode="none"),
|
||
)
|
||
|
||
assert state.active_workspace["candidate_set_binding"] == "cset_competitors_141"
|
||
assert state.context_capsules[0]["workspace_id"] == "ws_competitor_141"
|
||
assert rewritten.startswith("围绕当前研究工作区")
|
||
assert meta["mode"] == "active_workspace"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_state_scope_from_qa_record_prefers_customer_focus_object(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
qa = QARecord(
|
||
tenant_id=1,
|
||
question="现在有什么项目",
|
||
answer="你当前有 1 个项目：童析-学龄前儿童行为特质AI评估助手。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=["pmflow"],
|
||
response_time_ms=100,
|
||
asked_by=1,
|
||
asked_via="pmflow",
|
||
session_id=1,
|
||
turn_index=1,
|
||
routing_snapshot={
|
||
"response_payload": {
|
||
"customer_interaction_context_packet": {
|
||
"current_focus_object": {
|
||
"object_ref": "project:141",
|
||
"object_id": 141,
|
||
"object_type": "project",
|
||
"title": "童析-学龄前儿童行为特质AI评估助手",
|
||
}
|
||
}
|
||
}
|
||
},
|
||
)
|
||
|
||
scope = service._state_scope_from_qa_record(qa)
|
||
|
||
assert scope["project_id"] == 141
|
||
assert scope["target_ref"] == "project:141"
|
||
assert scope["target_entity_type"] == "project"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_short_reply_to_explanation_is_treated_as_meta_followup(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
db_session.add(
|
||
IngestEvent(
|
||
tenant_id=1,
|
||
channel="feishu_message",
|
||
event_key="om_user_message_explain_1",
|
||
status="done",
|
||
payload={"message_text": "你刚才说 adaptive_rag_answer_normal 调用失败。"},
|
||
)
|
||
)
|
||
db_session.add(
|
||
FeishuReplyOutbox(
|
||
tenant_id=1,
|
||
trace_id="trace-explain-1",
|
||
source_message_id="om_user_message_explain_1",
|
||
sent_message_id="om_bot_reply_explain_1",
|
||
chat_id="oc-chat-1",
|
||
message_text="系统处理失败：adaptive_rag_answer_normal LLM failed after retries: openai_chat_completion returned empty content after retries。"
|
||
"本次未执行入库。",
|
||
status="sent",
|
||
created_at=now - timedelta(minutes=1),
|
||
sent_at=now - timedelta(minutes=1),
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="这是什么情况，详细说下",
|
||
session_id=None,
|
||
context_session_id=None,
|
||
session_state=None,
|
||
thread_context={"reply_to_message_id": "om_bot_reply_explain_1"},
|
||
)
|
||
|
||
assert resolution.anchor_mode == "explicit"
|
||
assert resolution.is_meta_followup is True
|
||
assert resolution.meta_followup_kind == "reply_error_reason"
|
||
assert resolution.anchor_reply is not None
|
||
assert "你刚才说 adaptive_rag_answer_normal 调用失败" in resolution.anchor_reply["question"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_single_recent_reply_allows_short_explanation_followup_without_explicit_reply_to(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="错误追问",
|
||
external_conversation_key="feishu:chat-1:ou-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add(
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="你刚才说 adaptive_rag_answer_normal 调用失败",
|
||
answer="系统处理失败：adaptive_rag_answer_normal LLM failed after retries: openai_chat_completion returned empty content after retries。本次未执行入库。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=1200,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
source_message_id="om_user_error_1",
|
||
created_at=now,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="这是什么情况，详细说下",
|
||
session_id=int(session.id),
|
||
context_session_id=int(session.id),
|
||
session_state=state,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
)
|
||
|
||
assert resolution.anchor_mode == "implicit_recent_reply"
|
||
assert resolution.is_meta_followup is True
|
||
assert resolution.meta_followup_kind == "reply_error_reason"
|
||
assert resolution.anchor_reply is not None
|
||
assert resolution.anchor_reply["question"] == "你刚才说 adaptive_rag_answer_normal 调用失败"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_recent_followup_candidates_can_cross_sessions_by_conversation_scope_key(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
previous_session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="旧会话",
|
||
external_conversation_key="feishu:msg-old-1",
|
||
started_at=now - timedelta(minutes=2),
|
||
last_message_at=now - timedelta(minutes=1),
|
||
message_count=1,
|
||
)
|
||
current_session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="新会话",
|
||
external_conversation_key="feishu:msg-new-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add_all([previous_session, current_session])
|
||
await db_session.flush()
|
||
db_session.add(
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="帮我解释下系统架构",
|
||
answer="PMFlow 系统采用前后端分离架构。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=10,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(previous_session.id),
|
||
turn_index=1,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
source_message_id="om_user_1",
|
||
created_at=now - timedelta(minutes=1),
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(current_session.id))
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="这个是哪个模型回复的？",
|
||
session_id=int(current_session.id),
|
||
context_session_id=int(current_session.id),
|
||
session_state=state,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
)
|
||
|
||
assert resolution.anchor_mode == "implicit_recent_reply"
|
||
assert resolution.anchor_reply_id is not None
|
||
assert resolution.anchor_kind == "meta_recent_reply"
|
||
assert resolution.anchor_reply is not None
|
||
assert resolution.anchor_reply["question"] == "帮我解释下系统架构"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_standalone_project_lookup_is_not_misclassified_as_followup_anchor(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="旧问题",
|
||
external_conversation_key="feishu:chat-1:ou-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add(
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="帮我解释下系统架构",
|
||
answer="PMFlow 系统采用前后端分离架构。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=12,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
source_message_id="om_user_1",
|
||
created_at=now,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
resolution = await service._resolve_followup_anchor(
|
||
question="一个 AI 客服工具，这个项目有吗",
|
||
session_id=int(session.id),
|
||
context_session_id=int(session.id),
|
||
session_state=state,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
)
|
||
|
||
assert resolution.anchor_mode == "no_anchor"
|
||
assert resolution.anchor_reply_id is None
|
||
assert resolution.candidate_replies == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_state_rewrite_governance_skips_rewrite_when_current_scope_switches_project(
|
||
db_session,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="旧项目",
|
||
external_conversation_key="feishu:chat-1:ou-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add(
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？",
|
||
answer="先看测试执行结果、发布记录、工单反馈。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=12,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=81,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
routing_snapshot={
|
||
"query_semantics": {
|
||
"query_profile": {"referenced_objects": ["project_object"]},
|
||
"target_resolution": {
|
||
"target_entity_type": "project",
|
||
"target_ref": "project:81",
|
||
"resolved_label": "测试产能警告",
|
||
},
|
||
}
|
||
},
|
||
created_at=now,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
state.last_subject = "围绕“测试产能警告这个项目，如果只考虑外部信号源...”回答"
|
||
await db_session.flush()
|
||
|
||
async def _fake_query_semantics(**kwargs):
|
||
_ = kwargs
|
||
return {
|
||
"query_profile": {"referenced_objects": ["project_object"]},
|
||
"source_policy": "internal",
|
||
"effective_source_policy": "internal",
|
||
}
|
||
|
||
async def _fake_resolve_project_scope_for_question(**kwargs):
|
||
_ = kwargs
|
||
return (101, "AI客服", "AI客服")
|
||
|
||
monkeypatch.setattr(service, "_resolve_query_semantics", _fake_query_semantics, raising=False)
|
||
monkeypatch.setattr(
|
||
service,
|
||
"_resolve_project_scope_for_question",
|
||
_fake_resolve_project_scope_for_question,
|
||
raising=False,
|
||
)
|
||
|
||
rewrite_governance = await service._build_state_rewrite_governance(
|
||
question="AI客服这个项目里，用户洞察数据源现在主要是拿来解决什么问题？",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
source_policy_hint=None,
|
||
query_semantics=None,
|
||
session_state=state,
|
||
anchor_resolution=FollowupAnchorResolution(followup_mode="referential"),
|
||
)
|
||
|
||
assert rewrite_governance["allow_state_rewrite"] is False
|
||
assert rewrite_governance["reason"] == "new_project_scope"
|
||
assert rewrite_governance["current_scope"]["project_id"] == 101
|
||
assert rewrite_governance["previous_scope"]["project_id"] == 81
|
||
|
||
rewritten, rewrite_meta = service._rewrite_question_with_state(
|
||
question="AI客服这个项目里，用户洞察数据源现在主要是拿来解决什么问题？",
|
||
conversation=[],
|
||
session_state=state,
|
||
anchor_resolution=FollowupAnchorResolution(followup_mode="referential"),
|
||
rewrite_governance=rewrite_governance,
|
||
)
|
||
|
||
assert rewritten == "AI客服这个项目里，用户洞察数据源现在主要是拿来解决什么问题？"
|
||
assert rewrite_meta["mode"] == "scope_reset"
|
||
assert rewrite_meta["reason"] == "new_project_scope"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_state_rewrite_governance_skips_rewrite_when_current_scope_switches_business_target(
|
||
db_session,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="同项目旧业务",
|
||
external_conversation_key="feishu:chat-1:ou-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
db_session.add(
|
||
QARecord(
|
||
tenant_id=1,
|
||
question="测试产能警告这个项目里，告警接收人现在是谁？",
|
||
answer="当前还没定义明确接收人。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=12,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=81,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
routing_snapshot={
|
||
"query_semantics": {
|
||
"query_profile": {"referenced_objects": ["project_object", "signal_object"]},
|
||
"target_resolution": {
|
||
"target_entity_type": "signal",
|
||
"target_ref": "signal:alert_receiver",
|
||
"resolved_label": "告警接收人",
|
||
},
|
||
}
|
||
},
|
||
created_at=now,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
state = await service._ensure_session_state(session_id=int(session.id))
|
||
state.last_subject = "测试产能警告这个项目里，告警接收人现在是谁？"
|
||
await db_session.flush()
|
||
|
||
async def _fake_resolve_project_scope_for_question(**kwargs):
|
||
_ = kwargs
|
||
return (81, "测试产能警告", "测试产能警告")
|
||
|
||
monkeypatch.setattr(
|
||
service,
|
||
"_resolve_project_scope_for_question",
|
||
_fake_resolve_project_scope_for_question,
|
||
raising=False,
|
||
)
|
||
|
||
rewrite_governance = await service._build_state_rewrite_governance(
|
||
question="测试产能警告这个项目里，用户洞察数据源现在主要是拿来解决什么问题？",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
source_policy_hint=None,
|
||
query_semantics={
|
||
"query_profile": {"referenced_objects": ["project_object", "signal_object"]},
|
||
"target_resolution": {
|
||
"target_entity_type": "signal",
|
||
"target_ref": "signal:user_insight_source",
|
||
"resolved_label": "用户洞察数据源",
|
||
},
|
||
},
|
||
session_state=state,
|
||
anchor_resolution=FollowupAnchorResolution(followup_mode="referential"),
|
||
)
|
||
|
||
assert rewrite_governance["allow_state_rewrite"] is False
|
||
assert rewrite_governance["reason"] == "new_business_scope"
|
||
assert rewrite_governance["current_scope"]["project_id"] == 81
|
||
assert rewrite_governance["current_scope"]["target_ref"] == "signal:user_insight_source"
|
||
assert rewrite_governance["previous_scope"]["target_ref"] == "signal:alert_receiver"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_meta_followup_model_identity_reads_routing_snapshot_and_hides_real_model(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="系统架构",
|
||
external_conversation_key="feishu:chat-1:ou-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
qa = QARecord(
|
||
tenant_id=1,
|
||
question="帮我解释下系统架构",
|
||
answer="PMFlow 系统采用前后端分离架构。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=12,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
source_message_id="om_user_1",
|
||
routing_snapshot={
|
||
"planned_tier": "low",
|
||
"executed_tier": "high",
|
||
"route_source": "default",
|
||
"route_rule_name": "engine_high_reasoning_glm5",
|
||
"routing_decision": "low_to_high_escalated",
|
||
"provider": "openai",
|
||
"model": "GLM-5",
|
||
},
|
||
)
|
||
db_session.add(qa)
|
||
await db_session.commit()
|
||
|
||
payload = await service._build_meta_followup_payload(
|
||
question="这个是哪个模型回复的？",
|
||
anchor_resolution=FollowupAnchorResolution(
|
||
followup_mode="meta",
|
||
anchor_mode="implicit_recent_reply",
|
||
anchor_reply_id=int(qa.id),
|
||
anchor_session_id=int(session.id),
|
||
anchor_kind="meta_recent_reply",
|
||
is_meta_followup=True,
|
||
meta_followup_kind="reply_model_identity",
|
||
anchor_reply={
|
||
"qa_id": int(qa.id),
|
||
"question": "帮我解释下系统架构",
|
||
"anchor_reply_preview": "PMFlow 系统采用前后端分离架构。",
|
||
"source_message_id": "om_user_1",
|
||
},
|
||
),
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
)
|
||
|
||
assert payload is not None
|
||
assert payload["meta_followup_kind"] == "reply_model_identity"
|
||
assert payload["reason_code"] == "meta_followup_answered"
|
||
assert "planned_tier=low" in payload["answer"]
|
||
assert "executed_tier=high" in payload["answer"]
|
||
assert "engine_high_reasoning_glm5" in payload["answer"]
|
||
assert "glm-5" not in payload["answer"].lower()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_meta_followup_error_reason_humanizes_llm_empty_content_failure(db_session) -> None:
|
||
service = _Harness(db_session)
|
||
now = now_cn_naive()
|
||
session = QASession(
|
||
tenant_id=1,
|
||
user_id=None,
|
||
title="错误追问",
|
||
external_conversation_key="feishu:chat-1:ou-1",
|
||
started_at=now,
|
||
last_message_at=now,
|
||
message_count=1,
|
||
)
|
||
db_session.add(session)
|
||
await db_session.flush()
|
||
qa = QARecord(
|
||
tenant_id=1,
|
||
question="这是什么情况，详细说下",
|
||
answer="系统处理失败：adaptive_rag_answer_normal LLM failed after retries: openai_chat_completion returned empty content after retries。本次未执行入库。",
|
||
sources=[],
|
||
intent="question",
|
||
domains_used=[],
|
||
response_time_ms=1200,
|
||
asked_by=None,
|
||
asked_via="feishu",
|
||
project_id=None,
|
||
session_id=int(session.id),
|
||
turn_index=1,
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
source_chat_id="chat-1",
|
||
source_sender_open_id="ou-1",
|
||
source_message_id="om_user_error_1",
|
||
)
|
||
db_session.add(qa)
|
||
await db_session.commit()
|
||
|
||
payload = await service._build_meta_followup_payload(
|
||
question="这是什么情况，详细说下",
|
||
anchor_resolution=FollowupAnchorResolution(
|
||
anchor_mode="implicit_recent_reply",
|
||
anchor_reply_id=int(qa.id),
|
||
anchor_session_id=int(session.id),
|
||
anchor_kind="meta_recent_reply",
|
||
is_meta_followup=True,
|
||
meta_followup_kind="reply_error_reason",
|
||
anchor_reply={
|
||
"qa_id": int(qa.id),
|
||
"question": "你刚才说 adaptive_rag_answer_normal 调用失败",
|
||
"anchor_reply_preview": qa.answer,
|
||
"source_message_id": "om_user_error_1",
|
||
},
|
||
),
|
||
conversation_scope_key="feishu:chat-1:ou-1",
|
||
)
|
||
|
||
assert payload is not None
|
||
assert payload["meta_followup_kind"] == "reply_error_reason"
|
||
assert payload["reason_code"] == "meta_followup_answered"
|
||
assert "模型在生成答案时连续返回空正文" in payload["answer"]
|
||
assert "所以系统没有拿到可发出的有效回复" in payload["answer"]
|