brain.perception — Слой восприятия¶
Маршрутизация и нормализация входных данных. Аналог сенсорной коры.
InputRouter¶
Маршрутизатор входных данных: определяет тип (текст/файл/URL) и направляет к нужному ингестору.
InputRouter(text_ingestor: Optional[TextIngestor] = None, event_bus=None, dedup: bool = True, brain_logger: Optional[BrainLogger] = None, vision_ingestor: Optional[VisionIngestor] = None, audio_ingestor: Optional[AudioIngestor] = None)
Маршрутизатор входящих данных — аналог Таламуса.
MVP: обрабатывает только text-модальность. Image/audio/video — логируются как предупреждение и пропускаются.
Параметры
text_ingestor: экземпляр TextIngestor (создаётся автоматически) event_bus: EventBus для публикации событий (опционально) dedup: включить дедупликацию по SHA256 (default: True)
Использование
router = InputRouter()
Маршрутизация файла¶
events = router.route("docs/нейрон.pdf", session_id="s1")
Маршрутизация прямого текста¶
events = router.route_text("Нейрон — это клетка...", source="user_input")
Статистика¶
print(router.stats())
route(source: str, session_id: str = '', trace_id: str = '', force: bool = False, input_type: InputType = InputType.AUTO) -> List[PerceptEvent]
Маршрутизировать входящий источник (файл или строка текста).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
source
|
str
|
путь к файлу ИЛИ строка текста |
required |
session_id
|
str
|
ID сессии |
''
|
trace_id
|
str
|
ID трассировки |
''
|
force
|
bool
|
игнорировать дедупликацию (default: False) |
False
|
input_type
|
InputType
|
явный тип входных данных (FILE / TEXT / AUTO) |
AUTO
|
Returns:
| Type | Description |
|---|---|
List[PerceptEvent]
|
List[PerceptEvent] — пустой список при дубликате/reject/ошибке |
TextIngestor¶
Ингестор текстовых данных: нормализация, чанкинг, извлечение метаданных.
TextIngestor(chunk_min: int = CHUNK_MIN_CHARS, chunk_max: int = CHUNK_MAX_CHARS, overlap: int = CHUNK_OVERLAP, extractor: Optional[MetadataExtractor] = None)
Читает текстовые файлы и преобразует их в список PerceptEvent.
Параметры
chunk_min: минимальный размер чанка в символах (default: 1000) chunk_max: максимальный размер чанка в символах (default: 1500) overlap: перекрытие между чанками в символах (default: 120) extractor: экземпляр MetadataExtractor (создаётся автоматически)
Использование
ingestor = TextIngestor() events = ingestor.ingest("docs/нейрон.pdf")
→ List[PerceptEvent]¶
events = ingestor.ingest_text("Нейрон — это клетка...", source="user_input")
→ List[PerceptEvent]¶
Прочитать файл и вернуть список PerceptEvent (по одному на чанк).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_path
|
str
|
путь к файлу |
required |
session_id
|
str
|
ID сессии (прокидывается в события) |
''
|
trace_id
|
str
|
ID трассировки |
''
|
Returns:
| Type | Description |
|---|---|
List[PerceptEvent]
|
List[PerceptEvent] — пустой список при ошибке или hard reject |
Note
Метод не бросает исключений наружу. Ошибки логируются, а при неуспешной обработке возвращается пустой список.
MetadataExtractor¶
Извлечение метаданных из входных данных (язык, тип контента, источник).
Извлекает и вычисляет метаданные для текстового фрагмента.
Использование
extractor = MetadataExtractor() meta = extractor.extract( text="Нейрон — это клетка нервной системы...", source="docs/нейрон.pdf", file_path="docs/нейрон.pdf", page=1, chunk_id=0, )
meta["quality"] → 0.8¶
meta["language"] → "ru"¶
extract(text: str, source: str, file_path: Optional[str] = None, page: Optional[int] = None, chunk_id: int = 0, encoding: str = 'utf-8', extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]
Вычислить метаданные для текстового фрагмента.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
текст фрагмента |
required |
source
|
str
|
строка-идентификатор источника (путь, URL, "user_input") |
required |
file_path
|
Optional[str]
|
реальный путь к файлу (для file_size_kb) |
None
|
page
|
Optional[int]
|
номер страницы (для PDF) |
None
|
chunk_id
|
int
|
порядковый номер чанка в документе |
0
|
encoding
|
str
|
кодировка файла |
'utf-8'
|
extra
|
Optional[Dict[str, Any]]
|
дополнительные поля (будут добавлены в metadata) |
None
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
dict с полями: source, ts, quality, language, file_size_kb, encoding, page, chunk_id, warnings, ... |
staticmethod
¶
Определить язык текста по доле кириллических и латинских символов.
Делегирует в каноническую реализацию brain.core.text_utils.detect_language().
Returns:
| Type | Description |
|---|---|
str
|
'ru' | 'en' | 'mixed' | 'unknown' |
staticmethod
¶
Вычислить quality score по эвристикам из спецификации.
Компоненты
+0.3 длина > 50 символов +0.3 нет битых символов +0.2 язык определён (не 'unknown') +0.2 структура есть (есть переносы строк или несколько предложений)
Returns:
| Type | Description |
|---|---|
tuple[float, list[str]]
|
(quality: float 0.0–1.0, warnings: List[str]) |
staticmethod
¶
Вернуть метку качества по числовому значению.
Returns:
| Type | Description |
|---|---|
str
|
'normal' | 'warning' | 'low_priority' |
staticmethod
¶
Проверить, нужно ли жёстко отклонить контент (hard reject).
Hard reject только для
- пустой текст
- текст не извлёкся (None)
- слишком мало осмысленного текста (< 10 символов после strip)
Returns:
| Type | Description |
|---|---|
tuple[bool, str]
|
(reject: bool, reason: str) |
Validators¶
Валидация входных данных перед обработкой.
brain/perception/validators.py
Валидация файловых путей и размеров для Perception Layer.
Защита от
- Path traversal (../../../etc/passwd)
- Null bytes в пути ( )
- Symlink escape (ссылка за пределы allowed dirs)
- Слишком большие файлы (> MAX_FILE_SIZE_MB)
Использование
from brain.perception.validators import validate_file_path, check_file_size
safe, reason = validate_file_path("/some/path/file.txt") ok, size_mb = check_file_size("/some/path/file.txt")
module-attribute
¶
Максимальный размер файла в мегабайтах (по умолчанию).
Проверить безопасность файлового пути.
Проверки
- Null bytes в пути → reject
- Path traversal (.. компоненты) → reject
- Symlink escape (resolved path за пределами allowed_dirs) → reject
- Абсолютные пути к системным директориям → reject
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_path
|
str
|
путь к файлу (строка) |
required |
allowed_dirs
|
Optional[List[str]]
|
список разрешённых корневых директорий (опционально). Если None — проверяется только traversal и null bytes. |
None
|
Returns:
| Type | Description |
|---|---|
Tuple[bool, str]
|
(safe: bool, reason: str) — True если путь безопасен, иначе причина отказа. |
Проверить размер файла.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_path
|
str
|
путь к файлу |
required |
max_mb
|
float
|
максимальный размер в мегабайтах (default: MAX_FILE_SIZE_MB) |
MAX_FILE_SIZE_MB
|
Returns:
| Type | Description |
|---|---|
bool
|
(ok: bool, size_mb: float) — True если размер допустим. |
float
|
Если файл не существует, возвращает (False, 0.0). |