Private
Public Access
refactor(ai_client): extract _should_cache_gemini_result helper (Phase 10 site 4)
Site L1732: count_tokens block in _send_gemini had:
try: count_resp = _gemini_client.models.count_tokens(...)
... set should_cache based on total_tokens ...
except Exception as e: _append_comms('[COUNT FAILED]')
Per TIER1_REVIEW: logging is NOT a drain. MIGRATE to Result[bool].
New helper _should_cache_gemini_result(sys_instr: str) -> Result[bool]:
- Result(data=True) if token count >= 2048
- Result(data=False) if below threshold + [CACHING SKIPPED] comms note
- Result(data=False, errors=[ErrorInfo]) on SDK failure + [COUNT FAILED] comms
Caller: should_cache = _should_cache_gemini_result(sys_instr).data
Audit: ai_client BC 6 -> 5. Site L1732 (now shifted to L1752) no longer BC.
This commit is contained in:
+30
-10
@@ -1635,6 +1635,35 @@ def _delete_gemini_cache_result() -> Result[None]:
|
||||
errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to delete gemini cache: {e}", source="ai_client._delete_gemini_cache_result", original=e)],
|
||||
)
|
||||
|
||||
_GEMINI_CACHE_TOKEN_THRESHOLD: int = 2048
|
||||
|
||||
def _should_cache_gemini_result(sys_instr: str) -> Result[bool]:
|
||||
"""Decide whether the current Gemini context warrants caching.
|
||||
|
||||
Returns Result(data=True) if token count >= 2048, Result(data=False)
|
||||
if below threshold (with a [CACHING SKIPPED] comms note), or
|
||||
Result(data=False, errors=[ErrorInfo]) on SDK failure.
|
||||
|
||||
The caller (_send_gemini) ignores errors and treats failure as
|
||||
'do not cache' (safe default: cache create is expensive; skipping
|
||||
on count failure is a soft fallback to inline system_instruction).
|
||||
"""
|
||||
if _gemini_client is None:
|
||||
return Result(data=False)
|
||||
try:
|
||||
count_resp = _gemini_client.models.count_tokens(model=_model, contents=[sys_instr])
|
||||
total = count_resp.total_tokens
|
||||
if total and total >= _GEMINI_CACHE_TOKEN_THRESHOLD:
|
||||
return Result(data=True)
|
||||
_append_comms("OUT", "request", {"message": f"[CACHING SKIPPED] Context too small ({total} tokens < {_GEMINI_CACHE_TOKEN_THRESHOLD})"})
|
||||
return Result(data=False)
|
||||
except Exception as e:
|
||||
_append_comms("OUT", "request", {"message": f"[COUNT FAILED] {e}"})
|
||||
return Result(
|
||||
data=False,
|
||||
errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to count gemini tokens: {e}", source="ai_client._should_cache_gemini_result", original=e)],
|
||||
)
|
||||
|
||||
def _extract_gemini_thoughts(resp: Any) -> str:
|
||||
"""
|
||||
Extracts concatenated thinking text from a Gemini response object's parts.
|
||||
@@ -1721,16 +1750,7 @@ def _send_gemini(md_content: str, user_message: str, base_dir: str,
|
||||
safety_settings = [types.SafetySetting(category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH)]
|
||||
)
|
||||
|
||||
should_cache = False
|
||||
try:
|
||||
if _gemini_client:
|
||||
count_resp = _gemini_client.models.count_tokens(model=_model, contents=[sys_instr])
|
||||
if count_resp.total_tokens and count_resp.total_tokens >= 2048:
|
||||
should_cache = True
|
||||
else:
|
||||
_append_comms("OUT", "request", {"message": f"[CACHING SKIPPED] Context too small ({count_resp.total_tokens} tokens < 2048)"})
|
||||
except Exception as e:
|
||||
_append_comms("OUT", "request", {"message": f"[COUNT FAILED] {e}"})
|
||||
should_cache = _should_cache_gemini_result(sys_instr).data
|
||||
if should_cache and _gemini_client:
|
||||
try:
|
||||
_gemini_cache = _gemini_client.caches.create(
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
"""Phase 10 site 4: _should_cache_gemini_result helper."""
|
||||
import sys
|
||||
sys.path.insert(0, ".")
|
||||
|
||||
|
||||
def test_phase10_site4_should_cache_gemini_result_exists():
|
||||
import src.ai_client
|
||||
assert hasattr(src.ai_client, "_should_cache_gemini_result"), \
|
||||
"_should_cache_gemini_result helper missing"
|
||||
|
||||
|
||||
def test_phase10_site4_should_cache_gemini_result_returns_result():
|
||||
import src.ai_client
|
||||
import inspect
|
||||
fn = src.ai_client._should_cache_gemini_result
|
||||
sig = inspect.signature(fn)
|
||||
assert "Result" in str(sig.return_annotation), \
|
||||
f"_should_cache_gemini_result return must be Result, got {sig.return_annotation}"
|
||||
Reference in New Issue
Block a user