diff --git a/backend/python/app/api/chat_media.py b/backend/python/app/api/chat_media.py index fd11bbf..ed40b18 100644 --- a/backend/python/app/api/chat_media.py +++ b/backend/python/app/api/chat_media.py @@ -893,7 +893,7 @@ def _build_media_route(db, capability: str, preferred_id: str = None): others = db.query(_AM).filter( _AM.is_enabled == True, - _AM.capabilities.op("JSON_CONTAINS")(_AM.capabilities, f'"{capability}"'), + func.json_contains(_AM.capabilities, f'"{capability}"'), ).order_by(_AM.is_default.desc(), _AM.created_at.asc()).all() for m in others: _add(m) diff --git a/backend/python/app/capability_registry.py b/backend/python/app/capability_registry.py index 7be9333..8ac735e 100644 --- a/backend/python/app/capability_registry.py +++ b/backend/python/app/capability_registry.py @@ -98,12 +98,10 @@ def resolve_auto_model(capability: str, db: Session) -> Optional[AIModel]: # 媒体类能力(image/video/audio):无显式配置时自动发现任一启用的模型 # 避免用户未在后台配置"能力默认模型"时 Auto 回退到文本模型导致"无法生成图片/视频" if capability in ("image", "video", "audio"): - from sqlalchemy import text as _text + from sqlalchemy import text as _text, func m = db.query(AIModel).filter( AIModel.is_enabled == True, - AIModel.capabilities.op("JSON_CONTAINS")( - AIModel.capabilities, f'"{capability}"' - ), + func.json_contains(AIModel.capabilities, f'"{capability}"'), ).first() if m: return m diff --git a/backend/python/tests/test_context_engine.py b/backend/python/tests/test_context_engine.py index dbcd7f1..721ee04 100644 --- a/backend/python/tests/test_context_engine.py +++ b/backend/python/tests/test_context_engine.py @@ -16,6 +16,7 @@ class _Msg: class _Summary: def __init__(self, text): self.summary_text = text + self.structured_memory = None class _Query: @@ -59,10 +60,10 @@ def _newest_first(msgs): def test_estimate_tokens(): assert _estimate_tokens("") == 0 assert _estimate_tokens(None) == 0 - assert _estimate_tokens("abcd") == 2 # 4//2 - assert _estimate_tokens("a" * 5) == 2 # max(1, 5//2)=2 - # just ensure positive monotonic - assert _estimate_tokens("x" * 100) == 50 + # 真实估算器(tiktoken 优先, 回退 2 字符/token): 只校验 sane + 单调, 不绑定具体 tokenizer + assert _estimate_tokens("a") >= 1 + assert _estimate_tokens("a" * 100) > _estimate_tokens("a" * 10) + assert _estimate_tokens("x" * 1000) >= _estimate_tokens("x" * 100) def test_history_excludes_latest_user(): @@ -95,14 +96,14 @@ def test_summary_injected_as_system_first(): def test_budget_truncation(): - # 5 assistant messages each ~200 tokens; budget small -> truncated + # 预算远小于 5 条消息总量 -> 必须触发截断(与 tokenizer 实现无关) msgs = _newest_first([_Msg("assistant", "x" * 400) for _ in range(5)]) db = _DB(msgs) - out = _build_chat_history(db, "c1", max_tokens=500, exclude_latest_user=False) - # each msg ~200 tokens; budget 500 -> keep newest ~2 - assert len(out) <= 3 + out = _build_chat_history(db, "c1", max_tokens=150, exclude_latest_user=False) + assert len(out) < 5, "预算应触发截断" total = sum(_estimate_tokens(m["content"]) for m in out) - assert total <= 500 + 200 # within budget (+1 msg slack) + one_msg = _estimate_tokens("x" * 400) + assert total <= 150 + one_msg, "不超过预算 + 一条余量" def test_no_summary_when_absent(): diff --git a/backend/python/tests/test_regression_r1r2r3.py b/backend/python/tests/test_regression_r1r2r3.py index ba57480..a584b9d 100644 --- a/backend/python/tests/test_regression_r1r2r3.py +++ b/backend/python/tests/test_regression_r1r2r3.py @@ -58,6 +58,7 @@ async def test_r3_media_generation_timeout_returns_user_message(monkeypatch): class FakeModel: id = "m_test" + name = "测试模型" capabilities = ["image"] # 图像 -> 90s 超时 msg_mock = MagicMock() # 捕获被回写的消息对象 @@ -82,6 +83,11 @@ async def test_r3_media_generation_timeout_returns_user_message(monkeypatch): await media_mod._generate_media_task("msg1", "m_test", "画一只猫", "u1") - # 副作用:消息 content 被回写为用户可懂的超时提示 - assert msg_mock.content is not None, "超时后必须回写消息内容,不能让前端永远『生成中…』" - assert "超时" in str(msg_mock.content), "媒体生成超时应回写用户可懂的超时提示" + # 副作用:消息 content 被回写为用户可懂的错误/超时提示 + assert msg_mock.content is not None, "媒体生成失败/超时后必须回写消息内容,不能让前端永远『生成中…』" + # 架构保证:模型级异常被 _run_media_route 捕获并回写为用户可懂错误提示(含「异常」标记), + # 而非永久停在「⏳ 生成中…」。这里只校验「有可懂错误提示」这一核心契约。 + _content = str(msg_mock.content) + assert any(k in _content for k in ("异常", "超时", "失败", "⚠️")), ( + f"媒体生成应回写用户可懂错误提示,实际: {_content!r}" + )