QihuAI Design
d3d8d62b94
CI / gate (push) Has been cancelled
Real bug:
- capability_registry.resolve_auto_model and chat_media._build_media_route used
`Column.op("JSON_CONTAINS")(col, val)`, but SQLAlchemy's op() accepts only ONE
argument -> TypeError on every image/video/audio auto-route, silently failing
media generation (caught as "后台生成异常"). Replaced with func.json_contains().
Test reconciliation (no production behavior change):
- tests/test_context_engine.py: tiktoken upgrade made the hardcoded len//2 token
estimates wrong; relax to tokenizer-agnostic assertions and add
structured_memory=None to the _Summary mock (code now branches on it).
- tests/test_regression_r1r2r3.py: R3 FakeModel lacked `name` (code reads m.name);
also the assertion assumed a literal '超时' string, but the failover architecture
catches model-level exceptions and writes a user-understandable error marker
instead. Align the assertion to the real contract: a media failure/timeout must
be written back as a user-understandable error message, never left as '生成中…'.
CI gate: 28 backend pytest + 15 frontend vitest all green.
116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
"""Phase 2 context-engine regression: _build_chat_history budget/layering/summary.
|
|
|
|
Pure unit test with a fake DB session — no network, no real model. Guards the
|
|
layered-memory + compression + isolation behaviour against regressions.
|
|
"""
|
|
from app.api.chat_context import _estimate_tokens, _build_chat_history
|
|
|
|
|
|
class _Msg:
|
|
def __init__(self, role, content, cid="c1"):
|
|
self.role = role
|
|
self.content = content
|
|
self.conversation_id = cid
|
|
|
|
|
|
class _Summary:
|
|
def __init__(self, text):
|
|
self.summary_text = text
|
|
self.structured_memory = None
|
|
|
|
|
|
class _Query:
|
|
def __init__(self, rows):
|
|
self._rows = list(rows)
|
|
|
|
def filter(self, *a, **k):
|
|
return self
|
|
|
|
def order_by(self, *a, **k):
|
|
# test pre-sorts newest-first to match desc(created_at)
|
|
return self
|
|
|
|
def limit(self, n=None):
|
|
return self
|
|
|
|
def all(self):
|
|
return self._rows
|
|
|
|
def first(self):
|
|
return self._rows[0] if self._rows else None
|
|
|
|
|
|
class _DB:
|
|
def __init__(self, messages, summary=None):
|
|
self._messages = messages
|
|
self._summary = summary
|
|
|
|
def query(self, model):
|
|
if getattr(model, "__name__", "") == "ConversationSummary":
|
|
return _Query([self._summary] if self._summary else [])
|
|
return _Query(self._messages)
|
|
|
|
|
|
def _newest_first(msgs):
|
|
# _build_chat_history expects desc(created_at) -> newest first.
|
|
# Test authors write the list already newest-first, so return as-is.
|
|
return list(msgs)
|
|
|
|
|
|
def test_estimate_tokens():
|
|
assert _estimate_tokens("") == 0
|
|
assert _estimate_tokens(None) == 0
|
|
# 真实估算器(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():
|
|
# newest-first: [current user, prev assistant, older user]
|
|
msgs = _newest_first([
|
|
_Msg("user", "current question"),
|
|
_Msg("assistant", "previous answer"),
|
|
_Msg("user", "older question"),
|
|
])
|
|
db = _DB(msgs)
|
|
out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
|
|
roles = [m["role"] for m in out]
|
|
# current user dropped; chronological order = [older user, previous assistant]
|
|
assert roles == ["user", "assistant"]
|
|
assert out[0]["content"] == "older question"
|
|
|
|
|
|
def test_summary_injected_as_system_first():
|
|
msgs = _newest_first([
|
|
_Msg("user", "hi"),
|
|
_Msg("assistant", "hello"),
|
|
])
|
|
db = _DB(msgs, summary=_Summary("用户想做图片生成"))
|
|
out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
|
|
assert out[0]["role"] == "system"
|
|
assert out[0]["content"].startswith("【对话历史摘要")
|
|
assert "用户想做图片生成" in out[0]["content"]
|
|
# after summary: chronological history (assistant then user 'hi' excluded? exclude_latest_user drops 'hi')
|
|
assert out[1]["role"] == "assistant"
|
|
|
|
|
|
def test_budget_truncation():
|
|
# 预算远小于 5 条消息总量 -> 必须触发截断(与 tokenizer 实现无关)
|
|
msgs = _newest_first([_Msg("assistant", "x" * 400) for _ in range(5)])
|
|
db = _DB(msgs)
|
|
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)
|
|
one_msg = _estimate_tokens("x" * 400)
|
|
assert total <= 150 + one_msg, "不超过预算 + 一条余量"
|
|
|
|
|
|
def test_no_summary_when_absent():
|
|
msgs = _newest_first([_Msg("user", "hi"), _Msg("assistant", "hello")])
|
|
db = _DB(msgs)
|
|
out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
|
|
# 'hi' excluded; only assistant remains, no system summary
|
|
assert all(m["role"] != "system" for m in out)
|
|
assert [m["role"] for m in out] == ["assistant"]
|