From 0cbfb9f8548cfa0010bd8d7f6f3822e20b5a3cfb Mon Sep 17 00:00:00 2001 From: zhanghui Date: Fri, 24 Jul 2026 13:19:02 +0800 Subject: [PATCH] fix(observability): address OTel review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the merged OTel instrumentation (#352) surfaced four issues, all on the enabled path (default-off, so no impact until tracing is turned on): - search LLM client was unwrapped, so hybrid/agentic token usage — the heaviest LLM spend — never reached Langfuse. Wrap it with UsageRecordingClient when observability is enabled, mirroring get_llm_client(); graceful keyword-only degradation is preserved. - set_generation_usage overwrote token counts, undercounting any span that wraps more than one chat call (the now-wrapped agentic path). Accumulate instead of replacing. - recall_hit was emitted for uncalibrated methods (unbounded BM25 / single-route vector), a near-constant always-hit signal that inflates dashboards. Gate hit on calibrated methods (HYBRID/AGENTIC); keyword and vector emit only the raw top_score. - init_tracing / init_score_sink were not idempotent — a re-init without an intervening shutdown orphaned the export thread + OTLP socket + worker task. Tear down the previous instance first (init_score_sink is now async). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/everos/core/lifespan/tracing_lifespan.py | 2 +- .../core/observability/tracing/attributes.py | 15 +++- .../core/observability/tracing/provider.py | 5 ++ .../core/observability/tracing/scores.py | 26 ++++-- src/everos/memory/search/manager.py | 19 +++- src/everos/service/search.py | 13 ++- .../test_observability/test_scores.py | 36 +++++++- .../test_span_attributes.py | 14 +++ .../test_observability/test_tracing.py | 30 +++++++ .../test_memory/test_search/test_manager.py | 88 +++++++++++++++---- .../test_service/test_search_llm_client.py | 77 ++++++++++++++++ 11 files changed, 291 insertions(+), 34 deletions(-) create mode 100644 tests/unit/test_service/test_search_llm_client.py diff --git a/src/everos/core/lifespan/tracing_lifespan.py b/src/everos/core/lifespan/tracing_lifespan.py index 29fc4f7..70b1fde 100644 --- a/src/everos/core/lifespan/tracing_lifespan.py +++ b/src/everos/core/lifespan/tracing_lifespan.py @@ -40,7 +40,7 @@ class TracingLifespanProvider(LifespanProvider): """ settings = load_settings().observability enabled = init_tracing(settings) - scores = init_score_sink(settings) + scores = await init_score_sink(settings) logger.info("tracing_lifespan_startup", enabled=enabled, scores=scores) return enabled diff --git a/src/everos/core/observability/tracing/attributes.py b/src/everos/core/observability/tracing/attributes.py index 46d2f5a..d12d167 100644 --- a/src/everos/core/observability/tracing/attributes.py +++ b/src/everos/core/observability/tracing/attributes.py @@ -181,15 +181,26 @@ def set_generation_usage( ) -> None: """Record ``gen_ai.*`` model + token attributes on the current span. + Token counts ACCUMULATE: a span that wraps more than one ``chat`` call + (e.g. the agentic/rank search path issues several LLM calls under one + ``everos.search.rank`` span) sums each call's usage rather than letting + the last call overwrite the rest. ``model`` is set as-is (last wins). + No-op when OTel is absent or there is no active recording span, so LLM client wrappers can call it unconditionally. """ if not _OTEL_AVAILABLE: return span = _otel_trace.get_current_span() + # `.attributes` exists on a recording SDK span and reflects values set + # earlier in this span's life; a non-recording span has none (getattr + # falls back to {}), and set_attribute on it is itself a no-op. + existing = getattr(span, "attributes", None) or {} if model is not None: span.set_attribute(GEN_AI_REQUEST_MODEL, model) if input_tokens is not None: - span.set_attribute(GEN_AI_USAGE_INPUT_TOKENS, input_tokens) + prior = existing.get(GEN_AI_USAGE_INPUT_TOKENS, 0) + span.set_attribute(GEN_AI_USAGE_INPUT_TOKENS, prior + input_tokens) if output_tokens is not None: - span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, output_tokens) + prior = existing.get(GEN_AI_USAGE_OUTPUT_TOKENS, 0) + span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, prior + output_tokens) diff --git a/src/everos/core/observability/tracing/provider.py b/src/everos/core/observability/tracing/provider.py index 95630cb..e58d501 100644 --- a/src/everos/core/observability/tracing/provider.py +++ b/src/everos/core/observability/tracing/provider.py @@ -117,6 +117,11 @@ def init_tracing( logger.warning("observability_enabled_but_otel_not_installed") return False + # Idempotent: a re-init without an intervening shutdown would otherwise + # orphan the previous provider's export thread + OTLP socket. + if _provider is not None: + shutdown_tracing() + from everos import __version__ resource = Resource.create( diff --git a/src/everos/core/observability/tracing/scores.py b/src/everos/core/observability/tracing/scores.py index ac480eb..6256acb 100644 --- a/src/everos/core/observability/tracing/scores.py +++ b/src/everos/core/observability/tracing/scores.py @@ -117,11 +117,14 @@ class RecallScoreSink: _sink: RecallScoreSink | None = None -def init_score_sink(settings: ObservabilitySettings) -> bool: +async def init_score_sink(settings: ObservabilitySettings) -> bool: """Build + start the sink when Langfuse creds + emit_recall_scores are set. Returns True if a sink was installed, False otherwise (disabled, scores off, or missing creds) — in which case ``emit_recall_scores`` is a no-op. + + Idempotent: a re-init without an intervening shutdown tears down the + previous sink first, so its worker task + httpx client are not orphaned. """ global _sink if not settings.enabled or not settings.emit_recall_scores: @@ -132,6 +135,9 @@ def init_score_sink(settings: ObservabilitySettings) -> bool: if not (pk and sk and host): return False + if _sink is not None: + await shutdown_score_sink() + import httpx endpoint = host.rstrip("/") + "/api/public/scores" @@ -156,10 +162,15 @@ def emit_recall_scores( trace_id: str, observation_id: str | None, top_score: float, - hit: bool, + hit: bool | None, method: str, ) -> None: - """Enqueue recall_top_score + recall_hit; no-op when the sink is off.""" + """Enqueue recall_top_score (always) + recall_hit (only when ``hit`` is + a verdict); no-op when the sink is off. + + ``hit=None`` means the method's score is uncalibrated (KEYWORD/VECTOR), + so no hit verdict is pushed — only the raw top score. + """ if _sink is None: return comment = f"method={method}" @@ -168,11 +179,12 @@ def emit_recall_scores( trace_id, observation_id, "recall_top_score", float(top_score), comment ) ) - _sink.enqueue( - ScoreRecord( - trace_id, observation_id, "recall_hit", 1.0 if hit else 0.0, comment + if hit is not None: + _sink.enqueue( + ScoreRecord( + trace_id, observation_id, "recall_hit", 1.0 if hit else 0.0, comment + ) ) - ) async def shutdown_score_sink() -> None: diff --git a/src/everos/memory/search/manager.py b/src/everos/memory/search/manager.py index b06e046..85c8e0f 100644 --- a/src/everos/memory/search/manager.py +++ b/src/everos/memory/search/manager.py @@ -139,6 +139,14 @@ def _top_score(data: SearchData) -> float: return max((item.score for item in items), default=0.0) +# Methods whose top score is calibrated to a comparable [0, 1] scale +# (HYBRID → LR sigmoid, AGENTIC → cross-encoder), so the recall_hit +# threshold is meaningful. KEYWORD (unbounded BM25) and single-route +# VECTOR are excluded — a fixed threshold there yields a misleading +# near-constant "hit" that inflates cross-method dashboards. +_CALIBRATED_METHODS = frozenset({SearchMethod.HYBRID, SearchMethod.AGENTIC}) + + class SearchManager: """Orchestrates per-kind recall, fusion, and shape into the public DTO.""" @@ -224,11 +232,16 @@ class SearchManager: # Recall-quality signal on the span (always on; the Langfuse # scores push is separate and gated on creds — see the score sink). + # `hit` is only meaningful for calibrated-score methods; leave it + # unset for KEYWORD/VECTOR so an unbounded score isn't forced + # through a fixed threshold into a misleading always-hit verdict. top_score = _top_score(data) - threshold = load_settings().observability.recall_hit_threshold - hit = top_score >= threshold span.set_attribute("everos.search.top_score", top_score) - span.set_attribute("everos.search.hit", hit) + hit: bool | None = None + if req.method in _CALIBRATED_METHODS: + threshold = load_settings().observability.recall_hit_threshold + hit = top_score >= threshold + span.set_attribute("everos.search.hit", hit) # Push recall-quality scores to Langfuse out-of-band (no-op unless # a score sink is configured); attach to this retriever span. diff --git a/src/everos/service/search.py b/src/everos/service/search.py index e8369d6..cf9905f 100644 --- a/src/everos/service/search.py +++ b/src/everos/service/search.py @@ -105,7 +105,8 @@ def _get_llm_client() -> LLMClient | None: from everos.component.llm import build_llm_provider from everos.config import load_settings - cfg = load_settings().llm + settings = load_settings() + cfg = settings.llm if not cfg.api_key or not cfg.api_key.get_secret_value() or not cfg.base_url: logger.warning( "llm_not_configured", @@ -113,7 +114,15 @@ def _get_llm_client() -> LLMClient | None: ) _llm_client = None else: - _llm_client = build_llm_provider(cfg) + client = build_llm_provider(cfg) + # Record token usage for the hybrid/agentic path, mirroring + # get_llm_client() — otherwise the heaviest LLM spend (query + # decomposition + rerank judge) is invisible in Langfuse. + if settings.observability.enabled: + from everos.component.llm._usage_client import UsageRecordingClient + + client = UsageRecordingClient(client) + _llm_client = client logger.info("search_llm_built", model=cfg.model) _llm_resolved = True return _llm_client diff --git a/tests/unit/test_core/test_observability/test_scores.py b/tests/unit/test_core/test_observability/test_scores.py index 6cd80f7..40ff0c7 100644 --- a/tests/unit/test_core/test_observability/test_scores.py +++ b/tests/unit/test_core/test_observability/test_scores.py @@ -10,7 +10,41 @@ from __future__ import annotations import asyncio -from everos.core.observability.tracing.scores import RecallScoreSink, ScoreRecord +from pydantic import SecretStr + +from everos.config.settings import ObservabilitySettings +from everos.core.observability.tracing.scores import ( + RecallScoreSink, + ScoreRecord, + init_score_sink, + shutdown_score_sink, +) + + +async def test_init_score_sink_tears_down_previous() -> None: + """Re-init without an intervening shutdown must not leak the previous + sink's worker task + httpx client: the old sink is stopped first.""" + settings = ObservabilitySettings( + enabled=True, + emit_recall_scores=True, + langfuse_public_key="pk", + langfuse_secret_key=SecretStr("sk"), + langfuse_host="https://us.cloud.langfuse.com", + ) + from everos.core.observability.tracing import scores as scores_mod + + try: + assert await init_score_sink(settings) is True + first = scores_mod._sink + assert first is not None and first._task is not None + + assert await init_score_sink(settings) is True + # The previous sink was torn down (stop() nulls its task) and a new + # sink installed in its place. + assert first._task is None + assert scores_mod._sink is not first + finally: + await shutdown_score_sink() async def test_worker_sends_payload_in_langfuse_shape() -> None: diff --git a/tests/unit/test_core/test_observability/test_span_attributes.py b/tests/unit/test_core/test_observability/test_span_attributes.py index ee3a89f..dafe74d 100644 --- a/tests/unit/test_core/test_observability/test_span_attributes.py +++ b/tests/unit/test_core/test_observability/test_span_attributes.py @@ -83,6 +83,20 @@ def test_set_generation_usage_annotates_current_span( assert attrs["gen_ai.usage.output_tokens"] == 22 +def test_set_generation_usage_accumulates_across_calls( + captured: InMemorySpanExporter, +) -> None: + # A multi-call operation (e.g. agentic search issuing several chats) + # inside one span must SUM token usage, not overwrite with the last call. + with memory_span("everos.search.rank", observation_type="generation"): + set_generation_usage(model="gpt-x", input_tokens=10, output_tokens=5) + set_generation_usage(model="gpt-x", input_tokens=3, output_tokens=7) + force_flush() + attrs = captured.get_finished_spans()[0].attributes + assert attrs["gen_ai.usage.input_tokens"] == 13 + assert attrs["gen_ai.usage.output_tokens"] == 12 + + def test_set_generation_usage_outside_span_is_noop() -> None: # No active span → must not raise (and nothing to record). set_generation_usage(model="x", input_tokens=1, output_tokens=2) diff --git a/tests/unit/test_core/test_observability/test_tracing.py b/tests/unit/test_core/test_observability/test_tracing.py index 5e3518a..9431daa 100644 --- a/tests/unit/test_core/test_observability/test_tracing.py +++ b/tests/unit/test_core/test_observability/test_tracing.py @@ -125,6 +125,36 @@ async def test_child_spans_nest_across_asyncio_gather( assert spans[child].context.trace_id == trace_id +def test_init_tracing_tears_down_previous_provider() -> None: + """Re-init without an intervening shutdown must not leak the previous + provider (its export thread + OTLP socket): the old provider is shut + down first, so its span processor receives ``shutdown()``.""" + from opentelemetry.sdk.trace import SpanProcessor + + class _SpyProcessor(SpanProcessor): + def __init__(self) -> None: + self.shutdown_called = False + + def on_start(self, span: object, parent_context: object = None) -> None: + pass + + def on_end(self, span: object) -> None: + pass + + def shutdown(self) -> None: + self.shutdown_called = True + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return True + + settings = ObservabilitySettings(enabled=True, endpoint="http://collector.invalid") + first, second = _SpyProcessor(), _SpyProcessor() + init_tracing(settings, span_processor=first) + init_tracing(settings, span_processor=second) + assert first.shutdown_called is True + assert second.shutdown_called is False + + def test_resolve_otlp_target_derives_from_langfuse_creds() -> None: import base64 diff --git a/tests/unit/test_memory/test_search/test_manager.py b/tests/unit/test_memory/test_search/test_manager.py index b396d46..88ae50c 100644 --- a/tests/unit/test_memory/test_search/test_manager.py +++ b/tests/unit/test_memory/test_search/test_manager.py @@ -283,6 +283,36 @@ async def test_user_keyword_returns_episodes_only() -> None: assert resp.data.profiles == [] +async def test_recall_hit_emitted_only_for_calibrated_methods( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """recall_hit is meaningful only for calibrated-score methods (HYBRID LR + / AGENTIC rerank). For KEYWORD (unbounded BM25) the code must emit + top_score but NOT a hit verdict — a fixed 0.6 threshold against an + unbounded score is an always-hit signal that inflates the dashboard.""" + import everos.memory.search.manager as mgr_mod + + calls: list[dict[str, Any]] = [] + monkeypatch.setattr(mgr_mod, "current_trace_ids", lambda: ("t" * 32, "s" * 16)) + monkeypatch.setattr(mgr_mod, "emit_recall_scores", lambda **kw: calls.append(kw)) + + # KEYWORD: raw BM25 score well above the threshold, but uncalibrated. + kw_mgr = _build_manager(episode_sparse=[_episode_row("ep_1", score=7.5)]) + await kw_mgr.search(_user_req(method=SearchMethod.KEYWORD)) + assert len(calls) == 1 + assert calls[0]["method"] == "keyword" + assert calls[0]["top_score"] == 7.5 + assert calls[0]["hit"] is None # no hit verdict for an uncalibrated method + + # HYBRID: calibrated LR score → a real boolean hit verdict is emitted. + calls.clear() + hy_mgr = _build_manager(embedding=_StubEmbedding()) + await hy_mgr.search(_user_req(method=SearchMethod.HYBRID)) + assert len(calls) == 1 + assert calls[0]["method"] == "hybrid" + assert isinstance(calls[0]["hit"], bool) + + async def test_search_uses_propagated_request_id_when_bound() -> None: """When a request id is bound upstream (middleware), ``search`` reuses it instead of minting a fresh one, so the response id matches the trace.""" @@ -1062,41 +1092,63 @@ async def test_hybrid_agent_emits_recall_and_rank(_search_spans: Any) -> None: assert "everos.search.rank" in spans -async def test_search_emits_top_score_and_hit_on_span(_search_spans: Any) -> None: - """search sets everos.search.top_score (max item score) + everos.search.hit - (>= recall_hit_threshold, default 0.6) on the retriever span — always, no - Langfuse needed.""" +async def test_search_emits_top_score_without_hit_for_keyword( + _search_spans: Any, +) -> None: + """KEYWORD sets everos.search.top_score (max item score) but NOT + everos.search.hit — an unbounded BM25 score forced through a fixed + threshold would be a misleading always-hit verdict.""" mgr = _build_manager(episode_sparse=[_episode_row("ep_1", score=0.75)]) await mgr.search(_user_req(method=SearchMethod.KEYWORD)) - spans = _span_index(_search_spans) - attrs = spans["everos.memory.search"].attributes + attrs = _span_index(_search_spans)["everos.memory.search"].attributes assert attrs["everos.search.top_score"] == pytest.approx(0.75) + assert "everos.search.hit" not in attrs + + +async def test_search_emits_hit_when_calibrated_above_threshold( + _search_spans: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + """HYBRID is calibrated ([0, 1]) so a top score >= recall_hit_threshold + (default 0.6) sets everos.search.hit = True on the retriever span.""" + import everos.memory.search.manager as mgr_mod + + monkeypatch.setattr(mgr_mod, "_top_score", lambda data: 0.9) + mgr = _build_manager(embedding=_StubEmbedding()) + await mgr.search(_user_req(method=SearchMethod.HYBRID)) + attrs = _span_index(_search_spans)["everos.memory.search"].attributes + assert attrs["everos.search.top_score"] == pytest.approx(0.9) assert attrs["everos.search.hit"] is True -async def test_search_hit_false_when_below_threshold(_search_spans: Any) -> None: - mgr = _build_manager(episode_sparse=[_episode_row("ep_1", score=0.3)]) - await mgr.search(_user_req(method=SearchMethod.KEYWORD)) - spans = _span_index(_search_spans) - attrs = spans["everos.memory.search"].attributes +async def test_search_hit_false_when_calibrated_below_threshold( + _search_spans: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + import everos.memory.search.manager as mgr_mod + + monkeypatch.setattr(mgr_mod, "_top_score", lambda data: 0.3) + mgr = _build_manager(embedding=_StubEmbedding()) + await mgr.search(_user_req(method=SearchMethod.HYBRID)) + attrs = _span_index(_search_spans)["everos.memory.search"].attributes assert attrs["everos.search.top_score"] == pytest.approx(0.3) assert attrs["everos.search.hit"] is False -async def test_search_top_score_zero_when_no_results(_search_spans: Any) -> None: +async def test_search_top_score_zero_without_hit_for_keyword( + _search_spans: Any, +) -> None: mgr = _build_manager() # no candidates await mgr.search(_user_req(method=SearchMethod.KEYWORD)) - spans = _span_index(_search_spans) - attrs = spans["everos.memory.search"].attributes + attrs = _span_index(_search_spans)["everos.memory.search"].attributes assert attrs["everos.search.top_score"] == pytest.approx(0.0) - assert attrs["everos.search.hit"] is False + assert "everos.search.hit" not in attrs async def test_search_enqueues_recall_scores( _search_spans: Any, monkeypatch: pytest.MonkeyPatch ) -> None: - """When tracing is active, search hands recall_top_score/hit to the score - sink with the retriever span's trace_id (032x) + observation_id (016x).""" + """When tracing is active, search hands recall_top_score to the score + sink with the retriever span's trace_id (032x) + observation_id (016x). + KEYWORD is uncalibrated so hit is None (no recall_hit is pushed).""" import everos.memory.search.manager as mgr_mod captured: dict[str, Any] = {} @@ -1109,7 +1161,7 @@ async def test_search_enqueues_recall_scores( await mgr.search(_user_req(method=SearchMethod.KEYWORD)) assert captured["top_score"] == pytest.approx(0.75) - assert captured["hit"] is True + assert captured["hit"] is None assert captured["method"] == "keyword" assert len(captured["trace_id"]) == 32 assert len(captured["observation_id"]) == 16 diff --git a/tests/unit/test_service/test_search_llm_client.py b/tests/unit/test_service/test_search_llm_client.py new file mode 100644 index 0000000..ce748d5 --- /dev/null +++ b/tests/unit/test_service/test_search_llm_client.py @@ -0,0 +1,77 @@ +"""service.search._get_llm_client — token-usage wrapping parity with memorize. + +The hybrid/agentic search path is the most LLM-heavy flow (query +decomposition + refine + per-query fan-out + rerank judge). Its client +must be wrapped with ``UsageRecordingClient`` when observability is on, +exactly like ``memorize`` / the reflection strategies — otherwise the +biggest token spend is invisible in Langfuse. +""" + +from __future__ import annotations + +import importlib + +import pytest +from pydantic import SecretStr + +import everos.config as config_mod +from everos.component.llm._usage_client import UsageRecordingClient +from everos.config import Settings +from everos.config.settings import LLMSettings, ObservabilitySettings + +# `everos.service.search` the submodule is shadowed by the re-exported +# `search` function on the package, so resolve the module explicitly. +search_mod = importlib.import_module("everos.service.search") + + +def _reset(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(search_mod, "_llm_client", None, raising=False) + monkeypatch.setattr(search_mod, "_llm_resolved", False, raising=False) + + +def _patch_settings(monkeypatch: pytest.MonkeyPatch, *, enabled: bool) -> None: + cfg = Settings( + llm=LLMSettings( + model="gpt-4.1-mini", + api_key=SecretStr("sk-test"), + base_url="https://example.test", + ), + observability=ObservabilitySettings(enabled=enabled), + ) + monkeypatch.setattr(config_mod, "load_settings", lambda: cfg) + + +def test_search_llm_wrapped_when_observability_enabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _reset(monkeypatch) + _patch_settings(monkeypatch, enabled=True) + + client = search_mod._get_llm_client() + + assert isinstance(client, UsageRecordingClient) + + +def test_search_llm_not_wrapped_when_observability_disabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _reset(monkeypatch) + _patch_settings(monkeypatch, enabled=False) + + client = search_mod._get_llm_client() + + assert client is not None + assert not isinstance(client, UsageRecordingClient) + + +def test_search_llm_none_when_unconfigured(monkeypatch: pytest.MonkeyPatch) -> None: + # No credentials → graceful None (keyword-only degradation), regardless + # of observability. The wrapper must not change this contract. + _reset(monkeypatch) + cfg = Settings( + llm=LLMSettings(model="m", api_key=None, base_url=None), + observability=ObservabilitySettings(enabled=True), + ) + monkeypatch.setattr(config_mod, "load_settings", lambda: cfg) + + assert search_mod._get_llm_client() is None