tzt
51f1de2154
- T-G1 embedder.py：embed() 调 llama-server /v1/embeddings -> 对称 per-vector int8 量化（scale=127/max|v|，1024 维余弦扰动 ~1e-4 << 1e-2 验收）； 超时/连接/非200/空形状 -> EmbedderDown（D-G4 fail-closed）； /v1/embeddings OpenAI 兼容透传（双路由组：/sense 前缀 + /v1 无前缀， 不可用 503）；build_sense_routers 返回列表 - T-G2 observer.py：Observation + 批量缓冲 100ms 刷盘（asyncio.Queue + to_thread，D-P10）；三消费方共用；升级阶梯回写 outcome/executed_tier； D-G6 无 query 原文列（int8 BLOB + 特征 JSON） - 测试 +11（量化误差/范围/fail-closed×3/透传/单例/BLOB 持久化/批量刷盘/ 回写/无原文列），全量 381->386->381? 校正：386 passed - 待真机项：llama-server embedder 端点冒烟（需配置 embedder.base_url）
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
"""观察埋点测试（T-G2）：批量缓冲/刷盘/三消费方记录/升级回写。"""
|
|
import asyncio
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from gateway.sense.observer import Observation, Observer, get_observer, reset_observer
|
|
from gateway.sense.store import SenseStore
|
|
|
|
|
|
def test_observer_batch_flush(tmp_path):
|
|
"""log 入队 -> 100ms 内批量落盘；多条一次刷。"""
|
|
async def scenario():
|
|
store = SenseStore.init_db(tmp_path / "s.sqlite3")
|
|
reset_observer()
|
|
obs = Observer(store, flush_interval=0.05)
|
|
await obs.start()
|
|
for i in range(5):
|
|
obs.log(Observation(
|
|
request_id=f"r{i}", consumer="pipeline",
|
|
decided_tier="T2", executed_tier="T2",
|
|
probs={"t1": 0.2, "t2": 0.6, "t3": 0.2},
|
|
policy_version="v0-rule", features={"turns": 1}))
|
|
await asyncio.sleep(0.2)
|
|
await obs.stop()
|
|
return store
|
|
|
|
store = asyncio.run(scenario())
|
|
with store._lock, store._connect() as conn:
|
|
n = conn.execute("SELECT COUNT(*) AS c FROM tier_observations").fetchone()["c"]
|
|
assert n == 5
|
|
|
|
|
|
def test_observer_embedding_blob(tmp_path):
|
|
"""int8 BLOB 原样持久化（D-G6）。"""
|
|
async def scenario():
|
|
store = SenseStore.init_db(tmp_path / "s.sqlite3")
|
|
reset_observer()
|
|
obs = Observer(store, flush_interval=0.05)
|
|
await obs.start()
|
|
obs.log(Observation(request_id="rb", consumer="proxy",
|
|
decided_tier="T1", executed_tier="T1",
|
|
embedding=bytes([1, 127, 200]), policy_version="v0"))
|
|
await asyncio.sleep(0.2)
|
|
await obs.stop()
|
|
return store
|
|
|
|
store = asyncio.run(scenario())
|
|
with store._lock, store._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT embedding FROM tier_observations WHERE request_id='rb'").fetchone()
|
|
assert bytes(row["embedding"]) == bytes([1, 127, 200])
|
|
|
|
|
|
def test_get_observer_singleton(tmp_path):
|
|
"""同循环内 get_observer 返回单例；reset 后重建。"""
|
|
async def scenario():
|
|
store = SenseStore.init_db(tmp_path / "s.sqlite3")
|
|
reset_observer()
|
|
o1 = get_observer(store)
|
|
o2 = get_observer(store)
|
|
return o1 is o2
|
|
assert asyncio.run(scenario()) is True
|
|
|
|
|
|
def test_upgrade_outcome_writeback(tmp_path):
|
|
"""升级阶梯回写：outcome=escalated + executed_tier 更新。"""
|
|
store = SenseStore.init_db(tmp_path / "s.sqlite3")
|
|
store.insert_observation({
|
|
"request_id": "up1", "consumer": "pipeline", "decided_tier": "T1",
|
|
"executed_tier": "T1", "probs": "{}", "policy_version": "v0",
|
|
"features": "{}"})
|
|
assert store.update_outcome("up1", "escalated", executed_tier="T2") is True
|
|
with store._lock, store._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT outcome, executed_tier FROM tier_observations"
|
|
" WHERE request_id='up1'").fetchone()
|
|
assert row["outcome"] == "escalated" and row["executed_tier"] == "T2"
|
|
|
|
|
|
def test_no_query_text_persisted(tmp_path):
|
|
"""D-G6：观察表不含 query 原文（列结构断言）。"""
|
|
store = SenseStore.init_db(tmp_path / "s.sqlite3")
|
|
with store._lock, store._connect() as conn:
|
|
cols = [r["name"] for r in conn.execute(
|
|
"PRAGMA table_info(tier_observations)").fetchall()]
|
|
assert "query" not in cols and "query_text" not in cols
|
|
assert "embedding" in cols and "features" in cols
|