generated from kgod/ai-review-template
133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
"""B-Step1c: intent router settings, shadow logger, and Builder flow mount (P0 observe-only)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from app.chat_intent_classifier import RuleBasedChatIntentClassifier
|
|
from app.chat_intent_shadow import (
|
|
ChatIntentShadowLogger,
|
|
build_chat_intent_shadow,
|
|
)
|
|
from app.chat_intents import ChatTurnClassification
|
|
from app.settings import Settings, load_settings
|
|
from builder_flow_helpers import create_builder_session, send_message
|
|
|
|
|
|
def test_intent_router_settings_defaults() -> None:
|
|
settings = Settings()
|
|
assert settings.intent_router_mode == "off"
|
|
assert settings.intent_model is None
|
|
|
|
|
|
def test_intent_router_settings_from_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
|
|
monkeypatch.setenv("RESUME_AGENT_INTENT_ROUTER_MODE", "shadow")
|
|
monkeypatch.setenv("RESUME_AGENT_INTENT_MODEL", "kimi-k3")
|
|
settings = load_settings(tmp_path / "missing.env")
|
|
assert settings.intent_router_mode == "shadow"
|
|
assert settings.intent_model == "kimi-k3"
|
|
monkeypatch.setenv("RESUME_AGENT_INTENT_ROUTER_MODE", "bogus")
|
|
with pytest.raises(ValueError, match="INTENT_ROUTER_MODE"):
|
|
load_settings(tmp_path / "missing.env")
|
|
|
|
|
|
class _StubClassifier:
|
|
def __init__(self, result: ChatTurnClassification | None = None, exc: Exception | None = None) -> None:
|
|
self.result = result
|
|
self.exc = exc
|
|
|
|
def classify(self, message: str, *, state_summary: dict[str, Any]) -> ChatTurnClassification:
|
|
if self.exc:
|
|
raise self.exc
|
|
assert self.result is not None
|
|
return self.result
|
|
|
|
|
|
def _captured_events(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, dict[str, Any]]]:
|
|
events: list[tuple[str, dict[str, Any]]] = []
|
|
monkeypatch.setattr(
|
|
"app.chat_intent_shadow.log_ai_event",
|
|
lambda event, **fields: events.append((event, fields)),
|
|
)
|
|
return events
|
|
|
|
|
|
def test_shadow_logs_llm_vs_rule_disagreement(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
events = _captured_events(monkeypatch)
|
|
shadow = ChatIntentShadowLogger(
|
|
_StubClassifier(ChatTurnClassification(intent="ask_question", confidence=0.9)),
|
|
RuleBasedChatIntentClassifier(),
|
|
)
|
|
shadow.observe("没有", state_summary={})
|
|
assert events == [
|
|
(
|
|
"chat_intent_shadow",
|
|
{
|
|
"registry_version": "1",
|
|
"rule_intent": "no_info",
|
|
"llm_intent": "ask_question",
|
|
"llm_confidence": 0.9,
|
|
"disagreement": True,
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_shadow_logs_agreement_and_survives_llm_failure(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
events = _captured_events(monkeypatch)
|
|
agree = ChatIntentShadowLogger(
|
|
_StubClassifier(ChatTurnClassification(intent="no_info")),
|
|
RuleBasedChatIntentClassifier(),
|
|
)
|
|
agree.observe("没有", state_summary={})
|
|
assert events[0][1]["disagreement"] is False
|
|
|
|
failing = ChatIntentShadowLogger(
|
|
_StubClassifier(exc=RuntimeError("boom")),
|
|
RuleBasedChatIntentClassifier(),
|
|
)
|
|
failing.observe("没有", state_summary={}) # must not raise
|
|
assert events[1][0] == "chat_intent_shadow_error"
|
|
|
|
|
|
def test_build_chat_intent_shadow_requires_shadow_mode_and_openai() -> None:
|
|
openai = {"llm_provider": "openai", "openai_api_key": "k"}
|
|
assert build_chat_intent_shadow(Settings(**openai, intent_router_mode="off")) is None
|
|
assert build_chat_intent_shadow(Settings(llm_provider="rule", intent_router_mode="shadow")) is None
|
|
shadow = build_chat_intent_shadow(Settings(**openai, intent_router_mode="shadow"))
|
|
assert isinstance(shadow, ChatIntentShadowLogger)
|
|
tuned = build_chat_intent_shadow(Settings(**openai, intent_router_mode="shadow", intent_model="kimi-k3"))
|
|
assert tuned is not None
|
|
assert tuned.primary._client.settings.openai_model == "kimi-k3"
|
|
|
|
|
|
class _Spy:
|
|
def __init__(self, *, raises: bool = False) -> None:
|
|
self.raises = raises
|
|
self.calls: list[dict[str, Any]] = []
|
|
|
|
def observe(self, message: str, *, state_summary: dict[str, Any]) -> None:
|
|
if self.raises:
|
|
raise RuntimeError("spy boom")
|
|
self.calls.append({"message": message, "state_summary": state_summary})
|
|
|
|
|
|
def test_process_message_notifies_mounted_shadow(client: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
agent = client.app.state.resume_agent
|
|
spy = _Spy()
|
|
monkeypatch.setattr(agent, "_chat_intent_shadow", spy, raising=False)
|
|
session_id, _ = create_builder_session(client)
|
|
send_message(client, session_id, "新增一段教育经历")
|
|
assert spy.calls[0]["message"] == "新增一段教育经历"
|
|
assert "confirmed_entries" in spy.calls[0]["state_summary"]
|
|
|
|
|
|
def test_shadow_failure_never_breaks_routing(client: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
agent = client.app.state.resume_agent
|
|
monkeypatch.setattr(agent, "_chat_intent_shadow", _Spy(raises=True), raising=False)
|
|
session_id, _ = create_builder_session(client)
|
|
body = send_message(client, session_id, "新增一段教育经历")
|
|
assert body["turn"]["role"] == "assistant"
|