mirror of
https://github.com/vectorize-io/hindsight.git
synced 2026-09-14 19:31:49 +08:00
review(recall-perf): route the new knobs through config, tidy the timers, add tests
- HINDSIGHT_API_GZIP_MIN_SIZE / RECALL_DIAGNOSTIC_PHASES / LOOP_LAG_REPORT_SECONDS (renamed from LOOP_LAG) are HindsightConfig fields now, documented and in .env.example, instead of ad-hoc os.environ reads. - get_request_context: the timestamp line sat above the docstring, which demoted it to a no-op string. - _bind_bank_id decides once per function whether to time the recall body, and no longer swallows exceptions from the metrics call. - Reuse semaphore_wait_start / backend_acquire_start instead of parallel timers. - loop_lag: keep a strong reference to the probe task, drop the noqa lambda. - Tests: diagnostic-phase flag, TEI verify for http vs https, loop-lag probe. - ruff format (the verify-generated-files failure).
This commit is contained in:
Nicolò Boschi
@@ -192,6 +192,12 @@ HINDSIGHT_API_LLM_MODEL=gpt-4o-mini
|
||||
HINDSIGHT_API_HOST=0.0.0.0
|
||||
HINDSIGHT_API_PORT=8888
|
||||
HINDSIGHT_API_LOG_LEVEL=info
|
||||
# Min response bytes to gzip; negative disables compression (saves CPU on recall).
|
||||
# HINDSIGHT_API_GZIP_MIN_SIZE=1024
|
||||
# Diagnostic: log event-loop lag percentiles every N seconds. 0 disables.
|
||||
# HINDSIGHT_API_LOOP_LAG_REPORT_SECONDS=0
|
||||
# Record the diagnostic (subset) recall phase metrics.
|
||||
# HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES=true
|
||||
# Vocabulary used for every token count and chunk boundary (recall budgets, chunk
|
||||
# sizes, prompt fitting, embedding truncation). o200k_base matches current OpenAI
|
||||
# models; set cl100k_base to reproduce the counts Hindsight produced before this
|
||||
|
||||
@@ -4650,11 +4650,12 @@ def create_app(
|
||||
from hindsight_api.loop_lag import install as _install_loop_lag
|
||||
from hindsight_api.worker import WorkerPoller
|
||||
|
||||
config = get_config()
|
||||
|
||||
# Started here rather than at import time because it needs a running loop, and it must run
|
||||
# on the loop that actually serves requests — that is the only one whose lag says anything.
|
||||
_install_loop_lag()
|
||||
_install_loop_lag(config.loop_lag_report_seconds)
|
||||
|
||||
config = get_config()
|
||||
poller = None
|
||||
poller_task = None
|
||||
loop_watchdog = None
|
||||
@@ -4810,9 +4811,10 @@ def create_app(
|
||||
|
||||
# Compressing a recall response costs ~5% of the request's CPU. Tunable so a deployment
|
||||
# that is CPU-bound rather than bandwidth-bound can raise the floor past its response size.
|
||||
_gzip_min = int(os.environ.get("HINDSIGHT_API_GZIP_MIN_SIZE", "1024"))
|
||||
if _gzip_min >= 0:
|
||||
app.add_middleware(GZipMiddleware, minimum_size=_gzip_min)
|
||||
# A negative floor drops the middleware entirely.
|
||||
gzip_min_size = get_config().gzip_min_size
|
||||
if gzip_min_size >= 0:
|
||||
app.add_middleware(GZipMiddleware, minimum_size=gzip_min_size)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Patch OpenAPI schema: align ValidationError with Pydantic v2 error format
|
||||
@@ -4974,7 +4976,6 @@ def _register_routes(app: FastAPI):
|
||||
audited = _make_audited_http(lambda: getattr(app.state, "audit_logger", None))
|
||||
|
||||
def get_request_context(request: Request, authorization: str | None = Header(default=None)) -> RequestContext:
|
||||
request.scope.setdefault("hs_deps_t0", time.time())
|
||||
"""
|
||||
Extract request context from the Authorization header.
|
||||
|
||||
@@ -4989,6 +4990,8 @@ def _register_routes(app: FastAPI):
|
||||
empty by default, so no other header reaches extension code unless an
|
||||
operator opts in.
|
||||
"""
|
||||
# Dependency-resolution start, read by api_recall to split `http_to_handler`.
|
||||
request.scope.setdefault("hs_deps_t0", time.time())
|
||||
api_key = None
|
||||
if authorization:
|
||||
if authorization.lower().startswith("bearer "):
|
||||
@@ -5063,6 +5066,7 @@ def _register_routes(app: FastAPI):
|
||||
)
|
||||
|
||||
request.scope["hs_deps_done"] = time.time()
|
||||
|
||||
return _precheck_dep
|
||||
|
||||
# Global exception handler for authentication errors
|
||||
|
||||
@@ -627,6 +627,9 @@ ENV_GRAPH_RETRIEVER = "HINDSIGHT_API_GRAPH_RETRIEVER"
|
||||
ENV_RECALL_MAX_CONCURRENT = "HINDSIGHT_API_RECALL_MAX_CONCURRENT"
|
||||
ENV_RECALL_CONNECTION_BUDGET = "HINDSIGHT_API_RECALL_CONNECTION_BUDGET"
|
||||
ENV_RECALL_MAX_QUERY_TOKENS = "HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS"
|
||||
ENV_RECALL_DIAGNOSTIC_PHASES = "HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES"
|
||||
ENV_GZIP_MIN_SIZE = "HINDSIGHT_API_GZIP_MIN_SIZE"
|
||||
ENV_LOOP_LAG_REPORT_SECONDS = "HINDSIGHT_API_LOOP_LAG_REPORT_SECONDS"
|
||||
ENV_MENTAL_MODEL_REFRESH_CONCURRENCY = "HINDSIGHT_API_MENTAL_MODEL_REFRESH_CONCURRENCY"
|
||||
ENV_LINK_EXPANSION_PER_ENTITY_LIMIT = "HINDSIGHT_API_LINK_EXPANSION_PER_ENTITY_LIMIT"
|
||||
ENV_LINK_EXPANSION_TIMEOUT = "HINDSIGHT_API_LINK_EXPANSION_TIMEOUT"
|
||||
@@ -1408,6 +1411,9 @@ DEFAULT_GRAPH_RETRIEVER = "link_expansion"
|
||||
DEFAULT_RECALL_MAX_CONCURRENT = 32 # Max concurrent recall operations per worker
|
||||
DEFAULT_RECALL_CONNECTION_BUDGET = 4 # Max concurrent DB connections per recall operation
|
||||
DEFAULT_RECALL_MAX_QUERY_TOKENS = 500 # Maximum tokens allowed in recall query
|
||||
DEFAULT_RECALL_DIAGNOSTIC_PHASES = True # Record the subset (diagnostic=true) recall phase metrics
|
||||
DEFAULT_GZIP_MIN_SIZE = 1024 # Min response bytes to gzip; negative disables compression
|
||||
DEFAULT_LOOP_LAG_REPORT_SECONDS = 0.0 # Event-loop lag probe report interval; 0 disables it
|
||||
DEFAULT_MENTAL_MODEL_REFRESH_CONCURRENCY = 8 # Max concurrent mental model refreshes
|
||||
DEFAULT_LINK_EXPANSION_PER_ENTITY_LIMIT = 200 # Max target units per entity in graph expansion
|
||||
DEFAULT_LINK_EXPANSION_TIMEOUT = 10.0 # Timeout (seconds) for entity expansion query
|
||||
@@ -2956,6 +2962,9 @@ class HindsightConfig:
|
||||
recall_max_concurrent: int
|
||||
recall_connection_budget: int
|
||||
recall_max_query_tokens: int
|
||||
recall_diagnostic_phases: bool
|
||||
gzip_min_size: int
|
||||
loop_lag_report_seconds: float
|
||||
mental_model_refresh_concurrency: int
|
||||
link_expansion_per_entity_limit: int
|
||||
link_expansion_timeout: float
|
||||
@@ -4334,6 +4343,12 @@ class HindsightConfig:
|
||||
os.getenv(ENV_RECALL_CONNECTION_BUDGET, str(DEFAULT_RECALL_CONNECTION_BUDGET))
|
||||
),
|
||||
recall_max_query_tokens=int(os.getenv(ENV_RECALL_MAX_QUERY_TOKENS, str(DEFAULT_RECALL_MAX_QUERY_TOKENS))),
|
||||
recall_diagnostic_phases=os.getenv(
|
||||
ENV_RECALL_DIAGNOSTIC_PHASES, str(DEFAULT_RECALL_DIAGNOSTIC_PHASES)
|
||||
).lower()
|
||||
in ("true", "1", "yes"),
|
||||
gzip_min_size=int(os.getenv(ENV_GZIP_MIN_SIZE, str(DEFAULT_GZIP_MIN_SIZE))),
|
||||
loop_lag_report_seconds=float(os.getenv(ENV_LOOP_LAG_REPORT_SECONDS, str(DEFAULT_LOOP_LAG_REPORT_SECONDS))),
|
||||
mental_model_refresh_concurrency=int(
|
||||
os.getenv(ENV_MENTAL_MODEL_REFRESH_CONCURRENCY, str(DEFAULT_MENTAL_MODEL_REFRESH_CONCURRENCY))
|
||||
),
|
||||
|
||||
@@ -192,6 +192,9 @@ def _bind_bank_id(
|
||||
|
||||
def decorate(func: Callable[_P, Awaitable[_R]]) -> Callable[_P, Awaitable[_R]]:
|
||||
sig = inspect.signature(func)
|
||||
# Decided once per decorated function rather than per call: this decorator wraps every
|
||||
# bank-scoped engine method, and only recall has a phase breakdown to feed.
|
||||
times_recall_body = getattr(func, "__name__", None) == "recall_async"
|
||||
|
||||
@functools.wraps(func)
|
||||
async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
|
||||
@@ -208,12 +211,10 @@ def _bind_bank_id(
|
||||
return await func(*args, **kwargs)
|
||||
finally:
|
||||
_current_bank_id.reset(token)
|
||||
if func.__name__ == "recall_async":
|
||||
try:
|
||||
get_metrics_collector().record_recall_phase(
|
||||
"recall_async_body", time.time() - _t0_body, diagnostic=True)
|
||||
except Exception:
|
||||
pass
|
||||
if times_recall_body:
|
||||
get_metrics_collector().record_recall_phase(
|
||||
"recall_async_body", time.time() - _t0_body, diagnostic=True
|
||||
)
|
||||
|
||||
return wrapper
|
||||
|
||||
@@ -7292,7 +7293,7 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
_d = time.time() - _t0
|
||||
get_metrics_collector().record_recall_phase("validate_pre", _d)
|
||||
if _d > 0.100:
|
||||
logger.info('[RECALL PHASE] validate_pre=%.3fs bank=%s', _d, bank_id)
|
||||
logger.info("[RECALL PHASE] validate_pre=%.3fs bank=%s", _d, bank_id)
|
||||
if result:
|
||||
if result.tags is not None:
|
||||
tags = result.tags
|
||||
@@ -7308,7 +7309,7 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
_d = time.time() - _t0
|
||||
get_metrics_collector().record_recall_phase("fuzzy_tags", _d)
|
||||
if _d > 0.100:
|
||||
logger.info('[RECALL PHASE] fuzzy_tags=%.3fs bank=%s', _d, bank_id)
|
||||
logger.info("[RECALL PHASE] fuzzy_tags=%.3fs bank=%s", _d, bank_id)
|
||||
|
||||
# Map budget enum to thinking_budget number using bank-resolved config.
|
||||
# Function "fixed" preserves legacy {LOW: 100, MID: 300, HIGH: 1000}; function "adaptive"
|
||||
@@ -7318,7 +7319,7 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
_d = time.time() - _t0
|
||||
get_metrics_collector().record_recall_phase("bank_config", _d)
|
||||
if _d > 0.100:
|
||||
logger.info('[RECALL PHASE] bank_config=%.3fs bank=%s', _d, bank_id)
|
||||
logger.info("[RECALL PHASE] bank_config=%.3fs bank=%s", _d, bank_id)
|
||||
thinking_budget = _resolve_thinking_budget(budget_config_dict, budget, max_tokens)
|
||||
# Reranker candidate cap, optionally scaled by the same budget level (env-configured,
|
||||
# 0/unset → flat reranker_max_candidates). Static config, so read from get_config().
|
||||
@@ -7354,10 +7355,9 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
result = None
|
||||
error_msg = None
|
||||
semaphore_wait_start = time.time()
|
||||
_t0_sem = time.time()
|
||||
async with self._search_semaphore:
|
||||
get_metrics_collector().record_recall_phase("semaphore_acquire", time.time() - _t0_sem)
|
||||
semaphore_wait = time.time() - semaphore_wait_start
|
||||
get_metrics_collector().record_recall_phase("semaphore_acquire", semaphore_wait)
|
||||
# Retry loop for connection errors
|
||||
max_retries = 3
|
||||
for attempt in range(max_retries + 1):
|
||||
@@ -7396,7 +7396,9 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
enable_temporal_retrieval=enable_temporal_retrieval,
|
||||
enable_graph_retrieval=enable_graph_retrieval,
|
||||
)
|
||||
get_metrics_collector().record_recall_phase("search_with_retries", time.time() - _t0_swr2, diagnostic=True)
|
||||
get_metrics_collector().record_recall_phase(
|
||||
"search_with_retries", time.time() - _t0_swr2, diagnostic=True
|
||||
)
|
||||
break # Success - exit retry loop
|
||||
except OperationCancelledError:
|
||||
# Client disconnected — propagate to the HTTP layer (499);
|
||||
@@ -7448,7 +7450,7 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
_d = time.time() - _t0
|
||||
get_metrics_collector().record_recall_phase("validate_post", _d)
|
||||
if _d > 0.100:
|
||||
logger.info('[RECALL PHASE] validate_post=%.3fs bank=%s', _d, bank_id)
|
||||
logger.info("[RECALL PHASE] validate_post=%.3fs bank=%s", _d, bank_id)
|
||||
except Exception as hook_err:
|
||||
logger.warning(f"Post-recall hook error (non-fatal): {hook_err}")
|
||||
raise
|
||||
@@ -7607,7 +7609,6 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
tracer.start()
|
||||
|
||||
backend_acquire_start = time.time()
|
||||
_t0_swr = time.time()
|
||||
backend = await self._get_read_backend()
|
||||
tracer.add_phase_metric("backend_acquisition", time.time() - backend_acquire_start)
|
||||
recall_start = time.time()
|
||||
@@ -7637,7 +7638,7 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
embedding_span.set_attribute("hindsight.query", query[:100])
|
||||
|
||||
try:
|
||||
get_metrics_collector().record_recall_phase("swr_prelude", time.time() - _t0_swr)
|
||||
get_metrics_collector().record_recall_phase("swr_prelude", time.time() - backend_acquire_start)
|
||||
query_embeddings = await embedding_utils.generate_embeddings_batch(
|
||||
self.embeddings,
|
||||
[query],
|
||||
@@ -7757,9 +7758,7 @@ class MemoryEngine(MemoryEngineInterface):
|
||||
# serialization either side, and any time the request sat in the
|
||||
# channel. Recorded per-request because p99s of the individual stages
|
||||
# are not additive, so this gap cannot be derived after the fact.
|
||||
tracer.add_phase_metric(
|
||||
"store_hop_overhead", max(0.0, _full_elapsed - _store_reported)
|
||||
)
|
||||
tracer.add_phase_metric("store_hop_overhead", max(0.0, _full_elapsed - _store_reported))
|
||||
tracer.add_phase_metric(
|
||||
"full_recall",
|
||||
_full_elapsed,
|
||||
|
||||
@@ -12,7 +12,8 @@ other callbacks (or blocked in a synchronous call) while this one was ready. If
|
||||
requests are slow, the time is in a real await and the phases are missing one; if lag tracks
|
||||
request latency, the loop is oversubscribed and no amount of I/O tuning helps.
|
||||
|
||||
Enabled by HINDSIGHT_API_LOOP_LAG (seconds between reports); unset means the task never starts.
|
||||
Enabled by HINDSIGHT_API_LOOP_LAG_REPORT_SECONDS (seconds between reports); 0 means the task never
|
||||
starts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -28,16 +29,16 @@ logger = logging.getLogger(__name__)
|
||||
#: enough that the probe itself is not a meaningful share of the loop's work.
|
||||
_TICK_S = 0.05
|
||||
|
||||
#: Floor on the report interval, so a typo like `0.01` does not turn the probe into log spam.
|
||||
_MIN_REPORT_S = 1.0
|
||||
|
||||
def _interval() -> float | None:
|
||||
raw = os.getenv("HINDSIGHT_API_LOOP_LAG")
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
return max(1.0, float(raw))
|
||||
except ValueError:
|
||||
logger.warning("[loop-lag] ignoring unparseable HINDSIGHT_API_LOOP_LAG=%r", raw)
|
||||
return None
|
||||
# The loop only keeps a weak reference to a task, so an unreferenced one can be garbage-collected
|
||||
# mid-run and the probe would silently stop reporting.
|
||||
_tasks: set[asyncio.Task[None]] = set()
|
||||
|
||||
|
||||
def _percentile(sorted_lags: list[float], p: float) -> float:
|
||||
return sorted_lags[min(len(sorted_lags) - 1, int(len(sorted_lags) * p / 100))]
|
||||
|
||||
|
||||
async def _run(report_every: float) -> None:
|
||||
@@ -50,24 +51,24 @@ async def _run(report_every: float) -> None:
|
||||
await asyncio.sleep(_TICK_S)
|
||||
lags.append((time.monotonic() - t0 - _TICK_S) * 1000.0)
|
||||
lags.sort()
|
||||
n = len(lags)
|
||||
q = lambda p: lags[min(n - 1, int(n * p / 100))] # noqa: E731
|
||||
logger.info(
|
||||
"[loop-lag] pid=%d n=%d p50=%.1fms p90=%.1fms p99=%.1fms max=%.1fms",
|
||||
pid, n, q(50), q(90), q(99), lags[-1],
|
||||
pid,
|
||||
len(lags),
|
||||
_percentile(lags, 50),
|
||||
_percentile(lags, 90),
|
||||
_percentile(lags, 99),
|
||||
lags[-1],
|
||||
)
|
||||
|
||||
|
||||
def install() -> bool:
|
||||
"""Start the probe on the running loop. No-op unless the env var asks for it."""
|
||||
every = _interval()
|
||||
if every is None:
|
||||
return False
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
# Called before the loop exists (import time); the caller retries from a startup hook.
|
||||
return False
|
||||
asyncio.ensure_future(_run(every))
|
||||
logger.info("[loop-lag] armed: reporting every %.0fs", every)
|
||||
return True
|
||||
def install(report_every: float) -> asyncio.Task[None] | None:
|
||||
"""Start the probe on the running loop. No-op when `report_every` is 0 (the default)."""
|
||||
if report_every <= 0:
|
||||
return None
|
||||
report_every = max(_MIN_REPORT_S, report_every)
|
||||
task = asyncio.get_running_loop().create_task(_run(report_every))
|
||||
_tasks.add(task)
|
||||
task.add_done_callback(_tasks.discard)
|
||||
logger.info("[loop-lag] armed: reporting every %.0fs", report_every)
|
||||
return task
|
||||
|
||||
@@ -33,11 +33,6 @@ if TYPE_CHECKING:
|
||||
import asyncpg
|
||||
|
||||
|
||||
_RECALL_DIAGNOSTIC_PHASES = os.environ.get(
|
||||
"HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES", "true"
|
||||
).lower() not in ("0", "false", "no")
|
||||
|
||||
|
||||
def _get_tenant() -> str:
|
||||
"""Get current tenant (schema) from context for metrics labeling."""
|
||||
# Import here to avoid circular imports
|
||||
@@ -428,6 +423,7 @@ class MetricsCollector(MetricsCollectorBase):
|
||||
from .config import get_config
|
||||
|
||||
self._include_bank_id = get_config().metrics_include_bank_id
|
||||
self._record_diagnostic_phases = get_config().recall_diagnostic_phases
|
||||
|
||||
# Operation latency histogram (in seconds)
|
||||
# Records duration of retain, recall, reflect operations
|
||||
@@ -867,7 +863,7 @@ class MetricsCollector(MetricsCollectorBase):
|
||||
per-arm timing inside `parallel_retrieval`, say — so a consumer summing phases into a
|
||||
request total can exclude them instead of double-counting.
|
||||
"""
|
||||
if diagnostic and not _RECALL_DIAGNOSTIC_PHASES:
|
||||
if diagnostic and not self._record_diagnostic_phases:
|
||||
return
|
||||
attrs = {"phase": phase, "tenant": _get_tenant(), "diagnostic": str(bool(diagnostic)).lower()}
|
||||
# One instrument, not two: the histogram already carries `_count` for this attribute set,
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
"""The event-loop lag probe starts only when configured, and survives without a caller reference."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import pytest
|
||||
|
||||
from hindsight_api import loop_lag
|
||||
|
||||
|
||||
async def test_disabled_by_default() -> None:
|
||||
assert loop_lag.install(0) is None
|
||||
assert not loop_lag._tasks
|
||||
|
||||
|
||||
async def test_reports_lag_percentiles(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
|
||||
monkeypatch.setattr(loop_lag, "_MIN_REPORT_S", 0.0)
|
||||
monkeypatch.setattr(loop_lag, "_TICK_S", 0.001)
|
||||
caplog.set_level(logging.INFO, logger=loop_lag.__name__)
|
||||
|
||||
task = loop_lag.install(0.02)
|
||||
assert task is not None
|
||||
# Held by the module, so the loop's weak reference is not the only one keeping it alive.
|
||||
assert task in loop_lag._tasks
|
||||
try:
|
||||
for _ in range(100):
|
||||
if any("p99=" in r.getMessage() for r in caplog.records):
|
||||
break
|
||||
await asyncio.sleep(0.01)
|
||||
assert any("[loop-lag]" in r.getMessage() and "p99=" in r.getMessage() for r in caplog.records)
|
||||
finally:
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
await asyncio.sleep(0) # done callbacks run on the next loop iteration
|
||||
assert task not in loop_lag._tasks
|
||||
@@ -264,6 +264,24 @@ class TestMetricsCollector:
|
||||
attributes = collector.operation_duration.record.call_args[0][1]
|
||||
assert attributes["bank_id"] == "test_bank"
|
||||
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_recall_diagnostic_phases_follow_config(self, enabled):
|
||||
"""Diagnostic phases are dropped when disabled; ordinary phases are always recorded, once."""
|
||||
mock_config = MagicMock()
|
||||
mock_config.metrics_include_bank_id = False
|
||||
mock_config.recall_diagnostic_phases = enabled
|
||||
with (
|
||||
patch("hindsight_api.metrics.get_meter", return_value=MagicMock()),
|
||||
patch("hindsight_api.config.get_config", return_value=mock_config),
|
||||
):
|
||||
collector = MetricsCollector()
|
||||
|
||||
collector.record_recall_phase("engine_call", 0.01, diagnostic=True)
|
||||
collector.record_recall_phase("engine_auth", 0.01)
|
||||
|
||||
phases = [c.args[1]["phase"] for c in collector.recall_phase_duration.record.call_args_list]
|
||||
assert phases == (["engine_call", "engine_auth"] if enabled else ["engine_auth"])
|
||||
|
||||
|
||||
class TestGetMetricsCollector:
|
||||
"""Tests for the get_metrics_collector function."""
|
||||
|
||||
@@ -195,3 +195,26 @@ def test_tei_batch_size_env_var_reaches_the_client() -> None:
|
||||
else:
|
||||
os.environ[key] = value
|
||||
clear_config_cache()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("base_url", "expect_verify"),
|
||||
[("http://tei:8080", False), ("https://tei.example.com", True)],
|
||||
)
|
||||
def test_thread_client_skips_tls_setup_only_for_plaintext(
|
||||
monkeypatch: pytest.MonkeyPatch, base_url: str, expect_verify: bool
|
||||
) -> None:
|
||||
# A plaintext TEI never uses TLS, so building an SSLContext (and loading the CA bundle)
|
||||
# per thread client is pure overhead; an https TEI must still verify.
|
||||
seen: list[object] = []
|
||||
real_client = httpx.Client
|
||||
|
||||
def spy(*args: object, **kwargs: object) -> httpx.Client:
|
||||
seen.append(kwargs.get("verify", True))
|
||||
return real_client(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(httpx, "Client", spy)
|
||||
embeddings = RemoteTEIEmbeddings(base_url=base_url)
|
||||
embeddings._client_for_thread().close()
|
||||
|
||||
assert seen == [expect_verify]
|
||||
|
||||
@@ -1380,6 +1380,8 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust
|
||||
| `HINDSIGHT_API_LOG_FORMAT` | Log format: `text` or `json` (structured logging for cloud platforms) | `text` |
|
||||
| `HINDSIGHT_API_LOG_JSON_FIELDS` | Comma-separated allowlist of JSON log fields to emit (e.g. `severity,message,tenant`). Available: `severity`, `message`, `timestamp`, `logger`, `tenant`, `exception`. Empty = all fields. | `""` (all) |
|
||||
| `HINDSIGHT_API_MCP_ENABLED` | Enable MCP server at `/mcp/{bank_id}/` | `true` |
|
||||
| `HINDSIGHT_API_GZIP_MIN_SIZE` | Minimum response size (bytes) to gzip. Compressing a recall response costs ~5% of its CPU, so a CPU-bound (rather than bandwidth-bound) deployment can raise this past its typical response size. Negative disables compression entirely. | `1024` |
|
||||
| `HINDSIGHT_API_LOOP_LAG_REPORT_SECONDS` | Diagnostic: log event-loop lag percentiles (`[loop-lag]`) every N seconds (minimum 1), to tell a slow await from an oversubscribed loop. `0` disables the probe. | `0` |
|
||||
| `HINDSIGHT_API_TOKENIZER_ENCODING` | Vocabulary used for every token count and chunk boundary (recall budgets, chunk sizes, prompt fitting, embedding truncation). `o200k_base` matches current OpenAI models and counts non-Latin text far closer to what they actually charge; `cl100k_base` reproduces the counts Hindsight produced before this default changed. Server-level: token budgets are only comparable between banks if they are all counted the same way. Other bundled vocabulary: `o200k_harmony`. | `o200k_base` |
|
||||
| `HINDSIGHT_API_MODEL_INIT_TIMEOUT` | Wall-clock cap (seconds) on startup model/connection initialization. If embeddings, the cross-encoder, or LLM verification block (e.g. an offline model download or an unreachable provider), the server fails fast with a clear error instead of hanging forever. Increase if a legitimate first-time model download needs more time. | `300` |
|
||||
| `HINDSIGHT_API_STARTUP_WAIT_SECONDS` | **Docker image only.** How long the container waits for the API to answer `/health` before it stops and restarts. Raising `HINDSIGHT_API_MODEL_INIT_TIMEOUT` above the default raises this wait too, so a slow first-time model download is not cut short; set this to override the wait on its own. | `300`, or `HINDSIGHT_API_MODEL_INIT_TIMEOUT` + 30s when that is longer |
|
||||
@@ -2433,6 +2435,7 @@ Hindsight provides OpenTelemetry-based observability for LLM calls, conforming t
|
||||
| `HINDSIGHT_API_OTEL_SERVICE_NAME` | Service name for traces. Applies to the API and to standalone workers, which default to `hindsight-worker` when it is unset. | `hindsight-api` |
|
||||
| `HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT` | Deployment environment name (e.g., development, staging, production) | `development` |
|
||||
| `HINDSIGHT_API_METRICS_INCLUDE_BANK_ID` | Include `bank_id` in OTel metric attributes. Enable only for deployments with few banks — high cardinality causes unbounded memory growth. | `false` |
|
||||
| `HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES` | Record the diagnostic recall phases (`diagnostic="true"` on `hindsight.recall.phase.duration`) — subsets of other phases, useful only while diagnosing. Disable to cut instrument overhead on a busy recall path. | `true` |
|
||||
| `HINDSIGHT_API_METRICS_BACKLOG_ENABLED` | Expose async-operation queue depth and consolidation-backlog gauges (`hindsight_async_operations`, `hindsight_consolidation_backlog`, `hindsight_consolidation_failed`). Runs periodic per-schema `COUNT` queries on a background task. | `false` |
|
||||
| `OTEL_PYTHON_FASTAPI_EXCLUDED_URLS` | Comma-separated URL patterns excluded from request tracing | `health,metrics` |
|
||||
|
||||
|
||||
@@ -192,6 +192,12 @@ HINDSIGHT_API_LLM_MODEL=gpt-4o-mini
|
||||
HINDSIGHT_API_HOST=0.0.0.0
|
||||
HINDSIGHT_API_PORT=8888
|
||||
HINDSIGHT_API_LOG_LEVEL=info
|
||||
# Min response bytes to gzip; negative disables compression (saves CPU on recall).
|
||||
# HINDSIGHT_API_GZIP_MIN_SIZE=1024
|
||||
# Diagnostic: log event-loop lag percentiles every N seconds. 0 disables.
|
||||
# HINDSIGHT_API_LOOP_LAG_REPORT_SECONDS=0
|
||||
# Record the diagnostic (subset) recall phase metrics.
|
||||
# HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES=true
|
||||
# Vocabulary used for every token count and chunk boundary (recall budgets, chunk
|
||||
# sizes, prompt fitting, embedding truncation). o200k_base matches current OpenAI
|
||||
# models; set cl100k_base to reproduce the counts Hindsight produced before this
|
||||
|
||||
@@ -1380,6 +1380,8 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust
|
||||
| `HINDSIGHT_API_LOG_FORMAT` | Log format: `text` or `json` (structured logging for cloud platforms) | `text` |
|
||||
| `HINDSIGHT_API_LOG_JSON_FIELDS` | Comma-separated allowlist of JSON log fields to emit (e.g. `severity,message,tenant`). Available: `severity`, `message`, `timestamp`, `logger`, `tenant`, `exception`. Empty = all fields. | `""` (all) |
|
||||
| `HINDSIGHT_API_MCP_ENABLED` | Enable MCP server at `/mcp/{bank_id}/` | `true` |
|
||||
| `HINDSIGHT_API_GZIP_MIN_SIZE` | Minimum response size (bytes) to gzip. Compressing a recall response costs ~5% of its CPU, so a CPU-bound (rather than bandwidth-bound) deployment can raise this past its typical response size. Negative disables compression entirely. | `1024` |
|
||||
| `HINDSIGHT_API_LOOP_LAG_REPORT_SECONDS` | Diagnostic: log event-loop lag percentiles (`[loop-lag]`) every N seconds (minimum 1), to tell a slow await from an oversubscribed loop. `0` disables the probe. | `0` |
|
||||
| `HINDSIGHT_API_TOKENIZER_ENCODING` | Vocabulary used for every token count and chunk boundary (recall budgets, chunk sizes, prompt fitting, embedding truncation). `o200k_base` matches current OpenAI models and counts non-Latin text far closer to what they actually charge; `cl100k_base` reproduces the counts Hindsight produced before this default changed. Server-level: token budgets are only comparable between banks if they are all counted the same way. Other bundled vocabulary: `o200k_harmony`. | `o200k_base` |
|
||||
| `HINDSIGHT_API_MODEL_INIT_TIMEOUT` | Wall-clock cap (seconds) on startup model/connection initialization. If embeddings, the cross-encoder, or LLM verification block (e.g. an offline model download or an unreachable provider), the server fails fast with a clear error instead of hanging forever. Increase if a legitimate first-time model download needs more time. | `300` |
|
||||
| `HINDSIGHT_API_STARTUP_WAIT_SECONDS` | **Docker image only.** How long the container waits for the API to answer `/health` before it stops and restarts. Raising `HINDSIGHT_API_MODEL_INIT_TIMEOUT` above the default raises this wait too, so a slow first-time model download is not cut short; set this to override the wait on its own. | `300`, or `HINDSIGHT_API_MODEL_INIT_TIMEOUT` + 30s when that is longer |
|
||||
@@ -2433,6 +2435,7 @@ Hindsight provides OpenTelemetry-based observability for LLM calls, conforming t
|
||||
| `HINDSIGHT_API_OTEL_SERVICE_NAME` | Service name for traces. Applies to the API and to standalone workers, which default to `hindsight-worker` when it is unset. | `hindsight-api` |
|
||||
| `HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT` | Deployment environment name (e.g., development, staging, production) | `development` |
|
||||
| `HINDSIGHT_API_METRICS_INCLUDE_BANK_ID` | Include `bank_id` in OTel metric attributes. Enable only for deployments with few banks — high cardinality causes unbounded memory growth. | `false` |
|
||||
| `HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES` | Record the diagnostic recall phases (`diagnostic="true"` on `hindsight.recall.phase.duration`) — subsets of other phases, useful only while diagnosing. Disable to cut instrument overhead on a busy recall path. | `true` |
|
||||
| `HINDSIGHT_API_METRICS_BACKLOG_ENABLED` | Expose async-operation queue depth and consolidation-backlog gauges (`hindsight_async_operations`, `hindsight_consolidation_backlog`, `hindsight_consolidation_failed`). Runs periodic per-schema `COUNT` queries on a background task. | `false` |
|
||||
| `OTEL_PYTHON_FASTAPI_EXCLUDED_URLS` | Comma-separated URL patterns excluded from request tracing | `health,metrics` |
|
||||
|
||||
|
||||
Reference in New Issue
Block a user