Sadmin/vector
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

B2: tolerant JSON extraction from Claude responses + graceful fallback

Browse Source 
_extract_json_payload handles a json/JSON/bare fence, raw JSON and JSON
embedded in prose; any unparseable response or contract violation now
degrades to the deterministic scoring_service instead of raising.
Also guards the response envelope itself (content[0].text).

B3: single home for recommendation scoring

services/recommendation_service.py holds the rules; routers/stats.py and
backend/services/stats_service.py both delegate to it. Unified rules are the
union of the two old copies: same weights/threshold, substring matching
(superset of the old exact match), tolerant key aliases, health_flags rule
kept. Endpoint response contract unchanged.

Plus: Overpass circuit breaker and concurrent zone queries in geo_service -
128 sequential calls per analysis no longer each burn a connect timeout when
the host has no outbound network.

Tests: 152 -> 194 passed.
This commit is contained in:
root
2026-07-25 13:01:02 +00:00
parent 76278f5fe1
commit 8b3a2cbf7e
7 changed files with 509 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/routers/stats.py
+10 -13
View File
@@ -2,6 +2,7 @@ from fastapi import APIRouter
from backend.database import db
from backend.schemas import DashboardResponse, RecommendationRequest
from services.recommendation_service import score_recommendation
router = APIRouter(prefix='/api/v1/stats', tags=['stats'])
@@ -18,18 +19,14 @@ def stats_heatmap() -> dict:
@router.post('/recommendation')
def stats_recommendation(payload: RecommendationRequest) -> dict:
score = 0
if payload.age is not None and payload.age < 12:
score += 20
if payload.elapsed_hours is not None and payload.elapsed_hours >= 12:
score += 20
if payload.terrain_primary in {'лес', 'болото', 'вода'}:
score += 15
if payload.weather in {'дождь', 'туман', 'снег', 'ночь'}:
score += 15
if len(payload.health_flags) >= 2:
score += 15
result = score_recommendation(
age=payload.age,
elapsed_hours=payload.elapsed_hours,
terrain=payload.terrain_primary,
weather=payload.weather,
health_flags=payload.health_flags,
)
return {
'recommendation': 'Высокий приоритет на прочёс и дрон' if score >= 40 else 'Стандартный приоритет поиска',
'score': score,
'recommendation': result['recommendation'],
'score': result['score'],
}
backend/services/stats_service.py
+7 -25
View File
@@ -1,12 +1,17 @@
from __future__ import annotations
from typing import Any
try:
from backend.database import db
except ImportError: # pragma: no cover - compatibility for direct service imports
from database import db
# Scoring rules live in services/recommendation_service.py (B3) — this module
# only re-exports them so existing `stats_service.get_statistical_recommendation`
# callers keep working.
from services.recommendation_service import get_statistical_recommendation
__all__ = ['summary', 'heatmap', 'get_statistical_recommendation']
def summary() -> dict:
return db.stats()
@@ -14,26 +19,3 @@ def summary() -> dict:
def heatmap() -> list[dict]:
return db.heatmap()
def get_statistical_recommendation(case_data: dict[str, Any]) -> dict[str, Any]:
score = 0
age = case_data.get('age') or case_data.get('age_years')
elapsed = case_data.get('elapsed_hours')
terrain = str(case_data.get('terrain') or case_data.get('terrain_primary') or '').lower()
weather = str(case_data.get('weather') or case_data.get('precipitation') or '').lower()
if age is not None and age < 12:
score += 20
if elapsed is not None and elapsed >= 12:
score += 20
if any(token in terrain for token in ('лес', 'болото', 'вода')):
score += 15
if any(token in weather for token in ('дождь', 'туман', 'снег', 'ночь')):
score += 15
return {
'score': score,
'priority': 'high' if score >= 40 else 'normal',
'recommendation': 'Высокий приоритет на прочёс и дрон' if score >= 40 else 'Стандартный приоритет поиска',
}
backend/tests/test_claude_service.py
+132 -1
View File
@@ -11,7 +11,8 @@ from services.claude_service import (
analyze_case,
analyze_with_fallback,
AnalysisResult,
PrimaryZone
PrimaryZone,
_extract_json_payload
)
@@ -369,3 +370,133 @@ class TestImmediateActions:
result = await analyze_with_fallback(case_data)
assert any("10-15" in action or "камер" in action for action in result.immediate_actions)
class TestExtractJsonPayload:
"""Test tolerant JSON extraction from a model response (B2)."""
def test_bare_json(self):
assert _extract_json_payload('{"a": 1}') == {'a': 1}
def test_json_fence(self):
assert _extract_json_payload('```json\n{"a": 1}\n```') == {'a': 1}
def test_uppercase_json_fence(self):
assert _extract_json_payload('```JSON\n{"a": 1}\n```') == {'a': 1}
def test_bare_fence(self):
assert _extract_json_payload('```\n{"a": 1}\n```') == {'a': 1}
def test_prose_around_json(self):
content = 'Вот результат анализа:\n{"a": 1}\nНадеюсь, это поможет.'
assert _extract_json_payload(content) == {'a': 1}
def test_prose_around_fenced_json(self):
content = 'Разбор:\n```json\n{"a": 1}\n```\nКонец.'
assert _extract_json_payload(content) == {'a': 1}
def test_json_array_is_rejected(self):
"""A top-level array is not a valid AnalysisResult payload."""
with pytest.raises(ValueError):
_extract_json_payload('[1, 2, 3]')
def test_plain_text_raises(self):
with pytest.raises(ValueError):
_extract_json_payload('Извините, я не могу выполнить этот запрос.')
def test_empty_raises(self):
with pytest.raises(ValueError):
_extract_json_payload('')
def test_none_raises(self):
with pytest.raises(ValueError):
_extract_json_payload(None)
class TestClaudeResponseFallback:
"""Malformed Claude responses must degrade to scoring, not crash (B2)."""
def _mock_client(self, mock_client, text=None, payload=None):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = (
payload if payload is not None else {"content": [{"text": text}]}
)
mock_client.return_value.__aenter__.return_value.post = AsyncMock(
return_value=mock_response
)
@pytest.mark.asyncio
async def test_unparseable_text_falls_back(self):
"""Model answers in prose instead of JSON -> deterministic fallback."""
with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}):
with patch('httpx.AsyncClient') as mock_client:
self._mock_client(mock_client, text='Не могу помочь с этим.')
result = await analyze_case({'age': 10, 'terrain': 'лес'})
assert result.fallback_used is True
assert isinstance(result, AnalysisResult)
@pytest.mark.asyncio
async def test_truncated_json_falls_back(self):
with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}):
with patch('httpx.AsyncClient') as mock_client:
self._mock_client(mock_client, text='```json\n{"urgency": "высок')
result = await analyze_case({'age': 10})
assert result.fallback_used is True
@pytest.mark.asyncio
async def test_valid_json_missing_required_fields_falls_back(self):
"""Parseable JSON that violates the AnalysisResult contract."""
with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}):
with patch('httpx.AsyncClient') as mock_client:
self._mock_client(mock_client, text='{"urgency": "высокая"}')
result = await analyze_case({'age': 10})
assert result.fallback_used is True
@pytest.mark.asyncio
async def test_unexpected_envelope_falls_back(self):
"""API envelope without content[0].text -> fallback, not KeyError."""
with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}):
with patch('httpx.AsyncClient') as mock_client:
self._mock_client(mock_client, payload={'unexpected': 'shape'})
result = await analyze_case({'age': 10})
assert result.fallback_used is True
@pytest.mark.asyncio
async def test_unfenced_json_with_prose_succeeds(self):
"""Recovery path: valid payload wrapped in prose is still used."""
import json as _json
payload = {
"urgency": "высокая",
"primary_zones": [{
"priority": 1,
"name": "Лес север",
"direction": "N",
"distance": 1.5,
"reason": "Вероятное направление"
}],
"search_radius_km": 5.0,
"key_locations": ["водоёмы"],
"behavioral_prediction": "Движение по тропам",
"immediate_actions": ["Организовать поиск"],
"summary": "Резюме"
}
text = 'Результат:\n' + _json.dumps(payload, ensure_ascii=False) + '\nГотово.'
with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}):
with patch('httpx.AsyncClient') as mock_client:
self._mock_client(mock_client, text=text)
result = await analyze_case({'age': 10})
assert result.fallback_used is False
assert result.urgency == "высокая"
backend/tests/test_recommendation_service.py
+161
View File
@@ -0,0 +1,161 @@
"""
Tests for the unified recommendation scoring (B3).
Covers the shared scorer directly, the dict-based alias entry point, and the
/api/v1/stats/recommendation endpoint that now delegates to it.
"""
import pytest
from fastapi.testclient import TestClient
from backend.main import app
from services.recommendation_service import (
HIGH_PRIORITY_TEXT,
NORMAL_PRIORITY_TEXT,
get_statistical_recommendation,
score_recommendation,
)
client = TestClient(app)
class TestScoreRecommendation:
"""Weights and threshold of the shared scorer."""
def test_empty_case_scores_zero(self):
result = score_recommendation()
assert result['score'] == 0
assert result['priority'] == 'normal'
assert result['recommendation'] == NORMAL_PRIORITY_TEXT
def test_young_child(self):
assert score_recommendation(age=8)['score'] == 20
def test_age_at_threshold_not_counted(self):
assert score_recommendation(age=12)['score'] == 0
def test_long_elapsed(self):
assert score_recommendation(elapsed_hours=12)['score'] == 20
def test_short_elapsed_not_counted(self):
assert score_recommendation(elapsed_hours=11)['score'] == 0
def test_risky_terrain(self):
assert score_recommendation(terrain='лес')['score'] == 15
def test_adverse_weather(self):
assert score_recommendation(weather='дождь')['score'] == 15
def test_multiple_health_flags(self):
assert score_recommendation(health_flags=['эпилепсия', 'РАС'])['score'] == 15
def test_single_health_flag_not_counted(self):
assert score_recommendation(health_flags=['эпилепсия'])['score'] == 0
def test_terrain_matches_as_substring(self):
"""Substring match — the router previously required an exact match."""
assert score_recommendation(terrain='смешанный лес')['score'] == 15
def test_weather_matches_as_substring(self):
assert score_recommendation(weather='сильный дождь')['score'] == 15
def test_terrain_case_insensitive(self):
assert score_recommendation(terrain='ЛЕС')['score'] == 15
def test_unknown_terrain_scores_zero(self):
assert score_recommendation(terrain='поле')['score'] == 0
def test_high_priority_at_threshold(self):
result = score_recommendation(age=8, elapsed_hours=14)
assert result['score'] == 40
assert result['priority'] == 'high'
assert result['recommendation'] == HIGH_PRIORITY_TEXT
def test_just_below_threshold_is_normal(self):
result = score_recommendation(age=8, terrain='лес')
assert result['score'] == 35
assert result['priority'] == 'normal'
def test_all_factors(self):
result = score_recommendation(
age=6,
elapsed_hours=24,
terrain='болото',
weather='туман',
health_flags=['РАС', 'эпилепсия'],
)
assert result['score'] == 85
assert result['priority'] == 'high'
class TestGetStatisticalRecommendation:
"""Dict entry point and its field aliases."""
def test_age_alias(self):
assert get_statistical_recommendation({'age_years': 8})['score'] == 20
def test_age_preferred_over_alias(self):
assert get_statistical_recommendation({'age': 8, 'age_years': 30})['score'] == 20
def test_terrain_alias(self):
assert get_statistical_recommendation({'terrain_primary': 'лес'})['score'] == 15
def test_weather_alias(self):
assert get_statistical_recommendation({'precipitation': 'снег'})['score'] == 15
def test_health_flags_counted(self):
payload = {'health_flags': ['РАС', 'эпилепсия']}
assert get_statistical_recommendation(payload)['score'] == 15
def test_missing_keys_are_safe(self):
assert get_statistical_recommendation({})['score'] == 0
def test_none_values_are_safe(self):
payload = {'age': None, 'elapsed_hours': None, 'terrain': None, 'weather': None}
assert get_statistical_recommendation(payload)['score'] == 0
class TestRecommendationEndpoint:
"""The endpoint keeps its response contract while delegating."""
def test_returns_score_and_recommendation(self):
response = client.post(
'/api/v1/stats/recommendation',
json={'age': 8, 'elapsed_hours': 14},
)
assert response.status_code == 200
body = response.json()
assert body['score'] == 40
assert body['recommendation'] == HIGH_PRIORITY_TEXT
def test_normal_priority_case(self):
response = client.post('/api/v1/stats/recommendation', json={'age': 30})
assert response.status_code == 200
body = response.json()
assert body['score'] == 0
assert body['recommendation'] == NORMAL_PRIORITY_TEXT
def test_empty_payload_accepted(self):
response = client.post('/api/v1/stats/recommendation', json={})
assert response.status_code == 200
assert response.json()['score'] == 0
def test_endpoint_matches_shared_scorer(self):
payload = {
'age': 6,
'elapsed_hours': 24,
'terrain_primary': 'болото',
'weather': 'туман',
'health_flags': ['РАС', 'эпилепсия'],
}
response = client.post('/api/v1/stats/recommendation', json=payload)
assert response.status_code == 200
expected = score_recommendation(
age=payload['age'],
elapsed_hours=payload['elapsed_hours'],
terrain=payload['terrain_primary'],
weather=payload['weather'],
health_flags=payload['health_flags'],
)
assert response.json()['score'] == expected['score']
assert response.json()['recommendation'] == expected['recommendation']
services/claude_service.py
+47 -12
View File
@@ -171,21 +171,56 @@ async def analyze_with_claude(case_data: dict, api_key: str) -> AnalysisResult:
if response.status_code != 200:
raise Exception(f"Anthropic API error: {response.status_code} - {response.text}")
result = response.json()
content = result["content"][0]["text"]
try:
result = response.json()
content = result["content"][0]["text"]
analysis_data = _extract_json_payload(content)
analysis_data['fallback_used'] = False
return AnalysisResult(**analysis_data)
except Exception as e:
# Ответ модели пришёл в неожидаемом виде — не роняем анализ,
# а отдаём детерминированный результат scoring_service.
logger.warning(
f"Не удалось разобрать ответ Claude ({type(e).__name__}: {e}). "
"Используется fallback scoring service."
)
return await analyze_with_fallback(case_data)
# Парсим JSON из ответа
# Убираем возможные markdown блоки кода
if "```json" in content:
content = content.split("```json")[1].split("```")[0].strip()
elif "```" in content:
content = content.split("```")[1].split("```")[0].strip()
analysis_data = json.loads(content)
analysis_data['fallback_used'] = False
def _extract_json_payload(content: str) -> dict:
"""Extract a JSON object from a model response.
# Преобразуем в Pydantic модель
return AnalysisResult(**analysis_data)
Tolerates a ```json fence, a bare ``` fence, or raw JSON with
surrounding prose. Raises ValueError if nothing parseable is found.
"""
candidates = []
stripped = (content or '').strip()
if '```' in stripped:
for marker in ('```json', '```JSON', '```'):
if marker in stripped:
after = stripped.split(marker, 1)[1]
candidates.append(after.split('```', 1)[0].strip())
break
candidates.append(stripped)
# Last resort: the widest {...} span in the text.
start, end = stripped.find('{'), stripped.rfind('}')
if start != -1 and end > start:
candidates.append(stripped[start:end + 1])
for candidate in candidates:
if not candidate:
continue
try:
parsed = json.loads(candidate)
except (json.JSONDecodeError, TypeError):
continue
if isinstance(parsed, dict):
return parsed
raise ValueError('Не удалось извлечь JSON из ответа модели')
async def analyze_with_fallback(case_data: dict) -> AnalysisResult:
services/geo_service.py
+72 -8
View File
@@ -1,9 +1,12 @@
"""
Geo service for building search zones and querying OpenStreetMap data via Overpass API.
"""
import asyncio
import math
import json
import hashlib
import logging
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Optional, Tuple
@@ -26,6 +29,52 @@ CACHE_DIR = Path("/tmp/overpass_cache")
CACHE_TTL_HOURS = 24
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
logger = logging.getLogger(__name__)
# --- Overpass circuit breaker ---------------------------------------------
# build_search_zones issues ~128 Overpass calls per analysis. When the host has
# no outbound connectivity every one of them burns the full connect timeout,
# which turns a single analysis into several minutes of waiting for results
# that are empty anyway. After OVERPASS_FAILURE_THRESHOLD consecutive
# transport failures we stop calling out until OVERPASS_COOLDOWN_SECONDS have
# passed. Callers get the same {'elements': []} they already got on error.
OVERPASS_FAILURE_THRESHOLD = 3
OVERPASS_COOLDOWN_SECONDS = 60.0
OVERPASS_CONNECT_TIMEOUT = 5.0
_overpass_failures = 0
_overpass_open_until = 0.0
def _overpass_circuit_open() -> bool:
"""True while the breaker is tripped (skip network, return empty fast)."""
if _overpass_failures < OVERPASS_FAILURE_THRESHOLD:
return False
if time.monotonic() >= _overpass_open_until:
_reset_overpass_circuit()
return False
return True
def _record_overpass_failure() -> None:
global _overpass_failures, _overpass_open_until
_overpass_failures += 1
if _overpass_failures == OVERPASS_FAILURE_THRESHOLD:
_overpass_open_until = time.monotonic() + OVERPASS_COOLDOWN_SECONDS
logger.warning(
"Overpass API недоступен (%d подряд неудачных запросов). "
"Геоданные отключены на %.0f c, анализ продолжается без них.",
_overpass_failures,
OVERPASS_COOLDOWN_SECONDS,
)
def _reset_overpass_circuit() -> None:
global _overpass_failures, _overpass_open_until
_overpass_failures = 0
_overpass_open_until = 0.0
# Direction mappings
SEARCH_DISTANCES = [500, 1000, 2000, 5000]
@@ -167,8 +216,13 @@ async def query_overpass(query: str) -> Dict:
if cached is not None:
return cached
# Skip the network entirely while the breaker is tripped.
if _overpass_circuit_open():
return {'elements': []}
# Query API
async with httpx.AsyncClient(timeout=30.0) as client:
timeout = httpx.Timeout(30.0, connect=OVERPASS_CONNECT_TIMEOUT)
async with httpx.AsyncClient(timeout=timeout) as client:
try:
response = await client.post(
OVERPASS_URL,
@@ -181,8 +235,10 @@ async def query_overpass(query: str) -> Dict:
# Save to cache
save_to_cache(cache_key, data)
_reset_overpass_circuit()
return data
except Exception as e:
_record_overpass_failure()
# Return empty result on error
return {'elements': []}
@@ -296,8 +352,6 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in
);
out geom;
"""
roads_data = await query_overpass(roads_query)
roads_km = calculate_road_length(roads_data.get('elements', []))
# Query water bodies
water_query = f"""
@@ -308,8 +362,7 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in
);
out center;
"""
water_data = await query_overpass(water_query)
water_distance = find_nearest_distance(lat, lon, water_data.get('elements', []))
# Query settlements
settlement_query = f"""
@@ -319,8 +372,7 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in
);
out;
"""
settlement_data = await query_overpass(settlement_query)
settlement_distance = find_nearest_distance(lat, lon, settlement_data.get('elements', []))
# Query forests
forest_query = f"""
@@ -331,7 +383,19 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in
);
out geom;
"""
forest_data = await query_overpass(forest_query)
# The four queries are independent - issue them concurrently.
roads_data, water_data, settlement_data, forest_data = await asyncio.gather(
query_overpass(roads_query),
query_overpass(water_query),
query_overpass(settlement_query),
query_overpass(forest_query),
)
roads_km = calculate_road_length(roads_data.get('elements', []))
water_distance = find_nearest_distance(lat, lon, water_data.get('elements', []))
settlement_distance = find_nearest_distance(lat, lon, settlement_data.get('elements', []))
forest_pct = calculate_forest_coverage(forest_data.get('elements', []), radius_m)
# Calculate road density (km of roads per km²)
services/recommendation_service.py
+80
View File
@@ -0,0 +1,80 @@
"""
Statistical search-priority recommendation.
Single home for the recommendation scoring rules (B3). Previously duplicated
between `backend/routers/stats.py` (inline, exact-match on terrain/weather,
plus a health_flags rule) and `backend/services/stats_service.py` (substring
match on tolerant key aliases, no health_flags rule).
The unified rules below are the union of the two: identical weights and
threshold, substring matching (a superset of the old exact match), tolerant
input keys, and the health_flags rule kept. This is NOT the 7-factor zonal
scoring of `scoring_service` — it only labels a case high/normal priority.
Deliberately free of any database import so both the router and the service
layer can use it without side effects.
"""
from __future__ import annotations
from typing import Any
# Scoring weights and threshold — unchanged from both previous implementations.
WEIGHT_YOUNG_CHILD = 20
WEIGHT_LONG_ELAPSED = 20
WEIGHT_RISKY_TERRAIN = 15
WEIGHT_ADVERSE_WEATHER = 15
WEIGHT_MULTIPLE_HEALTH_FLAGS = 15
HIGH_PRIORITY_THRESHOLD = 40
YOUNG_CHILD_AGE = 12
LONG_ELAPSED_HOURS = 12
MULTIPLE_HEALTH_FLAGS = 2
RISKY_TERRAIN_TOKENS = ('лес', 'болото', 'вода')
ADVERSE_WEATHER_TOKENS = ('дождь', 'туман', 'снег', 'ночь')
HIGH_PRIORITY_TEXT = 'Высокий приоритет на прочёс и дрон'
NORMAL_PRIORITY_TEXT = 'Стандартный приоритет поиска'
def score_recommendation(
age: int | None = None,
elapsed_hours: int | None = None,
terrain: str | None = None,
weather: str | None = None,
health_flags: list[str] | None = None,
) -> dict[str, Any]:
"""Score a case and return {'score', 'priority', 'recommendation'}."""
score = 0
if age is not None and age < YOUNG_CHILD_AGE:
score += WEIGHT_YOUNG_CHILD
if elapsed_hours is not None and elapsed_hours >= LONG_ELAPSED_HOURS:
score += WEIGHT_LONG_ELAPSED
if any(token in str(terrain or '').lower() for token in RISKY_TERRAIN_TOKENS):
score += WEIGHT_RISKY_TERRAIN
if any(token in str(weather or '').lower() for token in ADVERSE_WEATHER_TOKENS):
score += WEIGHT_ADVERSE_WEATHER
if len(health_flags or []) >= MULTIPLE_HEALTH_FLAGS:
score += WEIGHT_MULTIPLE_HEALTH_FLAGS
is_high = score >= HIGH_PRIORITY_THRESHOLD
return {
'score': score,
'priority': 'high' if is_high else 'normal',
'recommendation': HIGH_PRIORITY_TEXT if is_high else NORMAL_PRIORITY_TEXT,
}
def get_statistical_recommendation(case_data: dict[str, Any]) -> dict[str, Any]:
"""Dict-based entry point, tolerant of the field aliases used across
the desktop / mobile / admin payloads."""
return score_recommendation(
age=case_data.get('age') if case_data.get('age') is not None else case_data.get('age_years'),
elapsed_hours=case_data.get('elapsed_hours'),
terrain=case_data.get('terrain') or case_data.get('terrain_primary'),
weather=case_data.get('weather') or case_data.get('precipitation'),
health_flags=case_data.get('health_flags'),
)