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

Import Vector lab project

Browse Source
This commit is contained in:
root
2026-06-06 18:31:55 +00:00
commit fcc22063ed
105 changed files with 31639 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/claude_service.py
+312
View File
@@ -0,0 +1,312 @@
"""
Claude AI service for case analysis with fallback to scoring service.
Integrates geo_service zones and scoring_service rankings.
"""
import os
import json
from typing import Dict, List, Optional
from pydantic import BaseModel
import httpx
from .geo_service import build_search_zones
from .scoring_service import WeightedScorer
class PrimaryZone(BaseModel):
priority: int
name: str
direction: str
distance: float
reason: str
class AnalysisResult(BaseModel):
urgency: str # "критическая", "высокая", "средняя", "низкая"
primary_zones: List[PrimaryZone]
search_radius_km: float
key_locations: List[str]
behavioral_prediction: str
immediate_actions: List[str]
summary: str
fallback_used: bool = False
async def analyze_case(case_data: dict) -> AnalysisResult:
"""
Анализирует данные случая с помощью Claude API с fallback на scoring_service.
Интегрирует:
- geo_service: построение зон поиска
- scoring_service: оценка и ранжирование зон
- Claude API: интеллектуальный анализ (если доступен)
Args:
case_data: Словарь с данными случая
- age: возраст
- gender: пол
- terrain: местность
- weather: погода
- time_missing: время пропажи
- last_location: последнее местоположение
- lat, lon: координаты (опционально)
- profiles: поведенческие профили (опционально)
- season: сезон (опционально)
Returns:
AnalysisResult: Структурированный результат анализа
"""
# Попытка использовать Claude API
api_key = os.getenv("ANTHROPIC_API_KEY")
if api_key:
try:
return await analyze_with_claude(case_data, api_key)
except Exception as e:
# Логируем ошибку и переходим на fallback
print(f"Claude API unavailable: {e}. Using fallback scoring service.")
# Fallback: используем только scoring_service
return await analyze_with_fallback(case_data)
async def analyze_with_claude(case_data: dict, api_key: str) -> AnalysisResult:
"""
Анализ с помощью Claude API с интеграцией geo и scoring сервисов.
"""
# Построить зоны поиска если есть координаты
zones_data = ""
if case_data.get('lat') and case_data.get('lon'):
try:
zones = await build_search_zones(
case_data['lat'],
case_data['lon'],
case_data
)
# Ранжировать зоны
scorer = WeightedScorer()
zones_dict = [
{
'direction': z.direction,
'distance_km': z.distance_km,
'forest_pct': z.forest_pct,
'road_density': z.road_density,
'water_distance_km': z.water_distance_km,
'settlement_distance_km': z.settlement_distance_km
}
for z in zones
]
ranked_zones = scorer.rank_zones(zones_dict, case_data)
# Топ-5 зон для промпта
top_zones = ranked_zones[:5]
zones_data = "\n\nТОП-5 ПРИОРИТЕТНЫХ ЗОН (по scoring_service):\n"
for zone in top_zones:
zones_data += f"- {zone['direction']} направление, {zone['distance_km']}км: "
zones_data += f"оценка {zone['score']}/100, "
zones_data += f"лес {zone['forest_pct']}%, "
zones_data += f"дороги {zone['road_density']} км/км²\n"
except Exception as e:
print(f"Geo/scoring service error: {e}")
# Формируем промпт на русском языке
prompt = f"""Ты — эксперт по поисково-спасательным операциям (ПСО) МЧС Республики Беларусь. Проанализируй следующий случай пропажи человека и дай структурированные рекомендации.
ДАННЫЕ СЛУЧАЯ:
- Возраст пропавшего: {case_data.get('age', 'не указан')} лет
- Пол: {case_data.get('gender', 'не указан')}
- Местность: {case_data.get('terrain', 'не указана')}
- Погодные условия: {case_data.get('weather', 'не указаны')}
- Время пропажи: {case_data.get('time_missing', 'не указано')}
- Последнее известное местоположение: {case_data.get('last_location', 'не указано')}
- Особые обстоятельства: {case_data.get('circumstances', 'нет')}
- Физическое состояние: {case_data.get('physical_condition', 'не указано')}
- Опыт нахождения на природе: {case_data.get('experience', 'не указан')}
- Поведенческие профили: {', '.join(case_data.get('profiles', [])) if case_data.get('profiles') else 'нет'}
- Сезон: {case_data.get('season', 'не указан')}{zones_data}
ЗАДАЧА:
Предоставь детальный анализ в формате JSON со следующими полями:
1. urgency: Уровень срочности ("критическая", "высокая", "средняя", "низкая")
2. primary_zones: Массив из 3-5 приоритетных зон поиска, каждая с полями:
- priority: номер приоритета (1 = самый высокий)
- name: название зоны (например "Ближний лес", "Водоём на севере")
- direction: направление от последней точки (N, NE, E, SE, S, SW, W, NW)
- distance: расстояние в км
- reason: обоснование выбора этой зоны
3. search_radius_km: Рекомендуемый радиус поиска в километрах
4. key_locations: Массив ключевых типов локаций для проверки (водоемы, дороги, постройки и т.д.)
5. behavioral_prediction: Прогноз поведения пропавшего на основе возраста и обстоятельств
6. immediate_actions: Массив немедленных действий, которые нужно предпринять
7. summary: Краткое резюме анализа (2-3 предложения)
ВАЖНО: Учитывай данные из scoring_service при формировании primary_zones. Отвечай ТОЛЬКО валидным JSON без дополнительного текста."""
# Вызываем Anthropic API
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
"https://api.anthropic.com/v1/messages",
headers={
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json"
},
json={
"model": "claude-sonnet-4-20250514",
"max_tokens": 4096,
"messages": [
{
"role": "user",
"content": prompt
}
]
}
)
if response.status_code != 200:
raise Exception(f"Anthropic API error: {response.status_code} - {response.text}")
result = response.json()
content = result["content"][0]["text"]
# Парсим 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
# Преобразуем в Pydantic модель
return AnalysisResult(**analysis_data)
async def analyze_with_fallback(case_data: dict) -> AnalysisResult:
"""
Fallback анализ используя только scoring_service без Claude API.
"""
# Определяем срочность на основе возраста и профилей
age = case_data.get('age', 10)
profiles = case_data.get('profiles', [])
if age <= 4 or 'эпилепсия' in profiles or 'РАС' in profiles:
urgency = "критическая"
elif age <= 7 or 'велосипед' in profiles:
urgency = "высокая"
elif age <= 12:
urgency = "средняя"
else:
urgency = "средняя"
# Построить зоны если есть координаты
primary_zones = []
search_radius_km = 5.0
if case_data.get('lat') and case_data.get('lon'):
try:
zones = await build_search_zones(
case_data['lat'],
case_data['lon'],
case_data
)
# Ранжировать зоны
scorer = WeightedScorer()
zones_dict = [
{
'direction': z.direction,
'distance_km': z.distance_km,
'forest_pct': z.forest_pct,
'road_density': z.road_density,
'water_distance_km': z.water_distance_km,
'settlement_distance_km': z.settlement_distance_km
}
for z in zones
]
ranked_zones = scorer.rank_zones(zones_dict, case_data)
# Топ-5 зон
for i, zone in enumerate(ranked_zones[:5]):
primary_zones.append(PrimaryZone(
priority=i + 1,
name=f"Зона {zone['direction']} {zone['distance_km']}км",
direction=zone['direction'],
distance=zone['distance_km'],
reason=f"Оценка {zone['score']}/100 по scoring_service"
))
# Радиус на основе дистанции
search_radius_km = scorer.distance_multiplier * 2.0
except Exception as e:
print(f"Geo/scoring service error in fallback: {e}")
# Если зоны не построены, используем базовые
if not primary_zones:
primary_zones = [
PrimaryZone(
priority=1,
name="Ближняя зона",
direction="N",
distance=0.5,
reason="Базовая зона поиска"
),
PrimaryZone(
priority=2,
name="Средняя зона",
direction="E",
distance=1.0,
reason="Расширенная зона поиска"
)
]
# Ключевые локации на основе возраста
if age <= 7:
key_locations = ["водоёмы", "укрытия", "густая растительность", "ближайшие постройки"]
elif age <= 12:
key_locations = ["водоёмы", "дороги", "тропы", "лесные массивы"]
else:
key_locations = ["дороги", "населённые пункты", "транспортные узлы", "водоёмы"]
# Поведенческий прогноз
if 'РАС' in profiles:
behavioral_prediction = "Высокий риск движения к водоёмам и ж/д путям. Может не откликаться на имя."
elif age <= 4:
behavioral_prediction = "Минимальное движение, вероятно находится близко к точке потери."
elif age <= 12:
behavioral_prediction = "Умеренное движение, может следовать по тропам или дорогам."
else:
behavioral_prediction = "Целенаправленное движение, возможен выход к населённым пунктам."
# Немедленные действия
immediate_actions = [
"Организовать поисковые группы",
"Проверить ближайшие водоёмы",
"Опросить свидетелей в районе последнего местоположения"
]
if 'РАС' in profiles:
immediate_actions.insert(0, "КРИТИЧНО: Перекрыть все водоёмы и ж/д пути в радиусе 5 км")
if 'велосипед' in profiles:
immediate_actions.insert(0, "Расширить зону поиска до 10-15 км, проверить дорожные камеры")
summary = f"Случай классифицирован как {urgency} срочность. "
summary += f"Рекомендуемый радиус поиска: {search_radius_km} км. "
summary += f"Приоритет: {key_locations[0]}."
return AnalysisResult(
urgency=urgency,
primary_zones=primary_zones,
search_radius_km=search_radius_km,
key_locations=key_locations,
behavioral_prediction=behavioral_prediction,
immediate_actions=immediate_actions,
summary=summary,
fallback_used=True
)