brain.cognition — Когнитивное ядро¶
Аналог префронтальной коры: координация всех когнитивных подсистем.
CognitiveCore¶
Главный orchestrator когнитивного цикла. Делегирует выполнение CognitivePipeline.
CognitiveCore(memory_manager: MemoryManagerProtocol, text_encoder: Optional[TextEncoderProtocol] = None, event_bus: Optional[EventBusProtocol] = None, resource_monitor: Optional[ResourceMonitorProtocol] = None, policy: Optional[PolicyConstraints] = None, llm_provider: Optional[LLMProvider] = None, brain_logger: Optional[BrainLogger] = None, trace_builder: Optional[TraceBuilder] = None, digest_gen: Optional[DigestGenerator] = None)
Orchestrator когнитивного ядра.
Координирует
- GoalManager — управление целями
- Planner — декомпозиция целей
- HypothesisEngine — генерация гипотез
- Reasoner — reasoning loop (Ring 1)
- ActionSelector — выбор действия
- RetrievalAdapter — структурированный retrieval
- ContradictionDetector — обнаружение противоречий
- UncertaintyMonitor — мониторинг тренда confidence
Зависимости (инъекция через конструктор): - memory_manager: для извлечения/сохранения фактов - text_encoder: для кодирования текста (опционально) - event_bus: для публикации событий (опционально) - resource_monitor: для проверки ресурсов (опционально)
Использование
core = CognitiveCore(memory_manager=mm) result = core.run("что такое нейрон?")
run(query: str, encoded_percept: Optional[EncodedPercept] = None, resources: Optional[Dict[str, Any]] = None, session_id: Optional[str] = None) -> CognitiveResult
Выполнить полный когнитивный цикл.
Делегирует выполнение CognitivePipeline (P3-10, Этап H + N + J). Цепочка из 17 явных шагов: create_context → auto_encode → get_resources → build_retrieval_query → create_goal → evaluate_salience → compute_budget → index_percept_vector → reason → detect_knowledge_gaps (Этап J) → llm_enhance (Этап N) → select_action (+PolicyLayer) → execute_action → complete_goal → build_result → publish_event → post_cycle (Этап J)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
текстовый запрос |
required |
encoded_percept
|
Optional[EncodedPercept]
|
закодированный перцепт (опционально) |
None
|
resources
|
Optional[Dict[str, Any]]
|
состояние ресурсов (опционально) |
None
|
session_id
|
Optional[str]
|
идентификатор сессии для связывания нескольких вызовов в один диалог. Если None — генерируется автоматически (каждый вызов = новая сессия). |
None
|
Возвращает CognitiveResult.
Опровергнуть факт — снизить confidence в SemanticMemory и удалить из векторного индекса при обнулении.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
concept
|
str
|
ключ понятия |
required |
delta
|
float
|
величина снижения confidence |
0.1
|
Удалить факт из SemanticMemory и векторного индекса.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
concept
|
str
|
ключ понятия |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True если факт был удалён, False если не найден |
Удалить элемент из векторного индекса.
Вызывается при удалении факта, обнулении confidence, или замене при разрешении противоречий.
CognitivePipeline¶
Явный 12-шаговый пайплайн когнитивного цикла (P3-10).
CognitivePipeline(memory: MemoryManagerProtocol, encoder: Optional[TextEncoderProtocol], event_bus: Optional[EventBusProtocol], resource_monitor: Optional[ResourceMonitorProtocol], policy: PolicyConstraints, goal_manager: GoalManager, reasoner: Reasoner, action_selector: ActionSelector, vector_backend: Optional[VectorRetrievalBackend], cycle_count_fn: Callable[[], int], llm_provider: Optional[LLMProvider] = None, brain_logger: Optional[BrainLogger] = None, trace_builder: Optional[TraceBuilder] = None, gap_detector: Optional[KnowledgeGapDetector] = None, online_learner: Optional[OnlineLearner] = None, boundary_guard: Optional[BoundaryGuard] = None, safety_policy: Optional[SafetyPolicyLayer] = None, audit_logger: Optional[AuditLogger] = None)
Явный пайплайн когнитивного цикла.
Каждый шаг — отдельный метод, принимающий CognitivePipelineContext. Шаги выполняются последовательно в методе run().
Для расширения — переопределить нужные step_* методы в подклассе.
Использование
pipeline = CognitivePipeline( memory=mm, encoder=enc, event_bus=bus, resource_monitor=rm, policy=policy, goal_manager=gm, reasoner=reasoner, action_selector=selector, vector_backend=vb, cycle_count_fn=lambda: core.cycle_count, ) result = pipeline.run("что такое нейрон?")
run(query: str, encoded_percept: Optional[EncodedPercept] = None, resources: Optional[Dict[str, Any]] = None, session_id: Optional[str] = None) -> CognitiveResult
Выполнить полный когнитивный цикл через явный пайплайн.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
текстовый запрос |
required |
encoded_percept
|
Optional[EncodedPercept]
|
закодированный перцепт (опционально) |
None
|
resources
|
Optional[Dict[str, Any]]
|
состояние ресурсов (опционально) |
None
|
session_id
|
Optional[str]
|
идентификатор сессии |
None
|
Returns:
| Type | Description |
|---|---|
CognitiveResult
|
CognitiveResult |
Шаг 1: Создать контекст (session_id, cycle_id, trace_id).
Шаг 2: Автоматическое кодирование запроса через TextEncoder.
Шаг 4: Получить состояние ресурсов.
Шаг 4: Обогатить запрос ключевыми словами из EncodedPercept.
Шаг 5: Определить тип цели и создать Goal.
Шаг 8: Индексировать вектор перцепта в VectorRetrievalBackend.
Шаг 9: Reasoning loop → ReasoningTrace.
Шаг 12: Выбор действия → ActionDecision (с PolicyLayer).
Порядок
- ActionSelector.select() → первичное решение
- PolicyLayer.apply_filters() → проверка допустимости
- Если действие заблокировано — override на первый допустимый
Шаг 15: Выполнить действие (LEARN → store + инкрементальная индексация).
Шаг 14: Завершить или провалить цель в GoalManager.
Шаг 15: Собрать CognitiveResult из всех компонентов.
Шаг 19: Публикация события через EventBus.
CognitivePipelineContext¶
Контекст, передаваемый между шагами пайплайна.
dataclass
¶
CognitivePipelineContext(query: str, session_id: Optional[str] = None, encoded_percept: Optional[EncodedPercept] = None, resources: Dict[str, Any] = dict(), retrieval_query: str = '', goal: Optional[Goal] = None, query_vector: Optional[list] = None, trace: Optional[ReasoningTrace] = None, decision: Optional[ActionDecision] = None, cognitive_context: Optional[CognitiveContext] = None, result: Optional[CognitiveResult] = None, salience: Optional[SalienceScore] = None, budget: Optional[AttentionBudget] = None, llm_enhanced: bool = False, llm_response_text: str = '', llm_provider_name: str = '', memory_search_result: Optional[Any] = None, knowledge_gap: Optional[Any] = None, safety_input_result: Optional[GuardResult] = None, safety_policy_result: Optional[SafetyDecision] = None, start_time: float = time.perf_counter(), elapsed_ms: float = 0.0, aborted: bool = False, abort_reason: str = '')
Контекст, передаваемый между шагами когнитивного пайплайна.
Каждый шаг читает нужные поля и записывает результаты. Поля заполняются последовательно по мере прохождения шагов.
GoalManager¶
Управление целями с приоритетной очередью.
Дерево целей с приоритетной очередью активных узлов.
Внутренняя структура
goal_tree: Dict[str, Goal] — все цели (id → Goal) active_queue: min-heap по (-priority, created_at, goal_id) interrupted_stack: List[Goal] — прерванные цели (LIFO) completed: List[str] — завершённые ID
Использование
gm = GoalManager() goal = Goal(description="ответить на вопрос", goal_type="answer_question") gm.push(goal) current = gm.peek() # → goal gm.complete(goal.goal_id)
dataclass
¶
Goal(goal_id: str = '', description: str = '', goal_type: str = 'answer_question', priority: float = 0.5, deadline: Optional[float] = None, parent_goal_id: Optional[str] = None, sub_goals: List[str] = list(), status: GoalStatus = GoalStatus.PENDING, created_at: str = '', context: Dict[str, Any] = dict(), trace_id: str = '', failure_reason: str = '')
Bases: ContractMixin
Цель когнитивного ядра.
goal_type определяет шаблон декомпозиции и stop conditions. Допустимые типы: "answer_question", "learn_fact", "verify_claim", "explore_topic", "plan".
classmethod
¶
Десериализация с str → GoalStatus.
Bases: str, Enum
Статусы цели.
Отдельный от TaskStatus, потому что цель и задача — разные сущности. У цели есть INTERRUPTED и CANCELLED, у задачи — нет.
Reasoner¶
Reasoning loop с гипотезами и итерациями.
Reasoner(memory_manager: MemoryManagerProtocol, hypothesis_engine: Optional[HypothesisEngine] = None, planner: Optional[Planner] = None, retrieval_adapter: Optional[RetrievalAdapter] = None, contradiction_detector: Optional[ContradictionDetector] = None, uncertainty_monitor: Optional[UncertaintyMonitor] = None)
Ring 1 reasoning loop
retrieve → detect contradictions → hypothesize → score → select.
Зависимости (инъекция через конструктор): - memory_manager: для извлечения фактов - hypothesis_engine: для генерации и оценки гипотез - planner: для check_stop_conditions - retrieval_adapter: для структурированного retrieval (optional) - contradiction_detector: для обнаружения противоречий (optional) - uncertainty_monitor: для мониторинга тренда confidence (optional)
Использование
reasoner = Reasoner( memory_manager=mm, hypothesis_engine=HypothesisEngine(), planner=Planner(), ) trace = reasoner.reason( query="что такое нейрон?", goal=goal, policy=PolicyConstraints(), resources={}, )
reason(query: str, goal: Goal, policy: Optional[PolicyConstraints] = None, resources: Optional[Dict[str, Any]] = None, query_vector: Optional[List[float]] = None) -> ReasoningTrace
Выполнить reasoning loop для запроса.
Цикл
- retrieve evidence
- generate hypotheses
- score hypotheses
- select best
- check stop conditions → loop or stop
Возвращает ReasoningTrace с полной информацией.
dataclass
¶
ReasoningTrace(trace_id: str = '', goal_id: str = '', query: str = '', steps: List[ReasoningStep] = list(), best_hypothesis_id: str = '', best_statement: str = '', hypothesis_count: int = 0, evidence_refs: List[str] = list(), outcome: str = '', stop_reason: str = '', total_iterations: int = 0, final_confidence: float = 0.0, total_duration_ms: float = 0.0, metadata: Dict[str, Any] = dict())
Bases: ContractMixin
Полный trace reasoning loop.
Расширен по сравнению с базовым TraceChain
- best_hypothesis_id: ID лучшей гипотезы
- best_statement: текст лучшей гипотезы
- hypothesis_count: сколько гипотез было сгенерировано
- evidence_refs: ID использованных доказательств
- outcome: CognitiveOutcome (код завершения)
- stop_reason: человекочитаемая причина остановки
- total_iterations: сколько итераций reasoning loop
- final_confidence: итоговая уверенность
dataclass
¶
ReasoningStep(step_id: str = '', step_type: str = '', description: str = '', duration_ms: float = 0.0, input_summary: str = '', output_summary: str = '', metadata: Dict[str, Any] = dict())
Bases: ContractMixin
Один шаг рассуждения в reasoning loop.
step_type: "retrieve" | "hypothesize" | "score" | "select" duration_ms: время выполнения шага
ActionSelector¶
Выбор действия на основе reasoning trace.
Выбор действия на основе ReasoningTrace и PolicyConstraints.
Логика выбора (приоритет сверху вниз): 1. goal_type == "learn_fact" → LEARN 2. outcome ∈ FAILURE_OUTCOMES → REFUSE или ASK_CLARIFICATION 3. confidence ≥ min_confidence → RESPOND_DIRECT 4. confidence ≥ min_confidence * 0.6 → RESPOND_HEDGED 5. hypothesis_count > 0 → ASK_CLARIFICATION 6. fallback → REFUSE
Использование
selector = ActionSelector() decision = selector.select( trace=reasoning_trace, goal_type="answer_question", policy=PolicyConstraints(), resources={}, )
select(trace: ReasoningTrace, goal_type: str = 'answer_question', policy: Optional[PolicyConstraints] = None, resources: Optional[Dict[str, Any]] = None) -> ActionDecision
Выбрать действие на основе результатов reasoning.
Возвращает ActionDecision.
dataclass
¶
ActionDecision(action: str = ActionType.REFUSE.value, statement: str = '', confidence: float = 0.0, reasoning: str = '', hypothesis_id: str = '', metadata: Dict[str, Any] = dict())
Bases: ContractMixin
Решение о действии.
action: тип действия (ActionType) statement: текст ответа / действия confidence: уверенность в решении [0..1] reasoning: краткое обоснование выбора hypothesis_id: ID гипотезы, на которой основано решение metadata: дополнительные данные
Bases: str, Enum
Типы действий когнитивного ядра.
RESPOND_DIRECT: прямой ответ (высокая уверенность) RESPOND_HEDGED: ответ с оговоркой (средняя уверенность) ASK_CLARIFICATION: запросить уточнение у пользователя REFUSE: отказ (нет данных / опасно / вне компетенции) LEARN: сохранить факт в память (explicit memory action)
LEARN ≠ Learning Loop (Stage I). LEARN — это осознанное действие "запомни факт", Learning Loop — автоматическое обучение из опыта.
Planner¶
Декомпозиция целей на шаги с 5 стратегиями реплана.
Декомпозиция целей на шаги + stop conditions + replan.
MVP: 4 шаблона декомпозиции. replan() = retry only.
Использование
planner = Planner() plan = planner.decompose(goal) outcome = planner.check_stop_conditions(state, limits, resources) retry_plan = planner.replan(failed_step, context, failure)
Декомпозировать цель на шаги по шаблону.
Если goal_type не найден в шаблонах — используется default.
check_stop_conditions(state: ReasoningState, limits: GoalTypeLimits, resources: Optional[Dict[str, Any]] = None) -> Optional[CognitiveOutcome]
Проверить stop conditions для текущего состояния reasoning loop.
Возвращает CognitiveOutcome если нужно остановиться, иначе None.
Порядок проверок
- resource_blocked — ресурсы исчерпаны
- step_limit_reached — превышен лимит итераций
- goal_completed — confidence ≥ threshold + стабильность
replan(failed_step: PlanStep, goal: Goal, failure: CognitiveOutcome, current_retry: int = 0, max_retries: int = 2, max_total_replans: int = 3, used_strategies: Optional[Set[ReplanStrategy]] = None, decompose_depth: int = 0) -> Optional[ExecutionPlan]
Попытка перепланирования после сбоя.
5 стратегий
RETRY: повторить тот же план NARROW_SCOPE: убрать explore шаги BROADEN_SCOPE: добавить дополнительный retrieve DECOMPOSE: разбить на 2 подплана (depth limit=1) ESCALATE: return None (отказ)
Защита от циклов
- max_total_replans=3 (глобальный ceiling)
- used_strategies: запрет повтора стратегии
- DECOMPOSE depth limit=1
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
failed_step
|
PlanStep
|
шаг, на котором произошёл сбой |
required |
goal
|
Goal
|
цель |
required |
failure
|
CognitiveOutcome
|
тип сбоя |
required |
current_retry
|
int
|
текущий номер повтора |
0
|
max_retries
|
int
|
максимум повторов (legacy, respected) |
2
|
max_total_replans
|
int
|
глобальный лимит перепланирований |
3
|
used_strategies
|
Optional[Set[ReplanStrategy]]
|
уже использованные стратегии |
None
|
decompose_depth
|
int
|
текущая глубина DECOMPOSE |
0
|
Returns:
| Type | Description |
|---|---|
Optional[ExecutionPlan]
|
Новый ExecutionPlan или None (отказ) |
dataclass
¶
ExecutionPlan(plan_id: str = '', goal_id: str = '', steps: List[PlanStep] = list(), is_retry: bool = False, retry_count: int = 0)
Bases: ContractMixin
План выполнения цели.
steps: упорядоченный список шагов goal_id: ID цели, для которой создан план plan_id: уникальный ID плана is_retry: True если план создан через replan() retry_count: номер повтора (0 = первый план)
dataclass
¶
PlanStep(step_id: str = '', step_type: str = '', description: str = '', params: Dict[str, Any] = dict(), completed: bool = False, result: Optional[Any] = None)
Bases: ContractMixin
Один шаг плана выполнения цели.
step_type определяет, что делать: "retrieve" — извлечь факты из памяти "hypothesize"— сгенерировать гипотезы "score" — оценить гипотезы "select" — выбрать лучшую гипотезу "act" — выбрать действие (respond/learn/ask/refuse) "store" — сохранить факт в память
HypothesisEngine¶
Генерация и оценка гипотез (4 стратегии + бюджет).
Движок гипотез: генерация, оценка и ранжирование.
4 стратегии
- associative: прямое сопоставление по ключевым словам
- deductive: логический вывод из нескольких evidence
- causal: причинно-следственные связи (temporal/causal markers)
- analogical: кросс-доменные аналогии (разные memory_type/concept_refs)
Budget
- max_hypotheses_total = 3 (глобальный лимит)
- max_per_strategy = 2 (лимит на стратегию)
- dedup по normalized statement
Использование
engine = HypothesisEngine() hypotheses = engine.generate(query="что такое нейрон?", evidence=[...]) scored = engine.score_all(hypotheses, evidence) ranked = engine.rank(scored)
Сгенерировать гипотезы на основе запроса и доказательств.
Стратегии (в порядке приоритета): 1. Associative — прямое сопоставление 2. Deductive — логический вывод из нескольких evidence 3. Causal — причинно-следственные связи 4. Analogical — кросс-доменные аналогии
Budget: max_per_strategy=2, max_hypotheses_total=3, dedup.
Оценить одну гипотезу по доказательствам.
support - risk
support = Σ(relevance * confidence) для связанных evidence risk = Σ(contradiction_count * 0.2) для связанных evidence
Возвращает final_score (может быть отрицательным).
Оценить все гипотезы. Возвращает тот же список (мутирует).
Ранжировать гипотезы по final_score (desc). Stable sort: при равном score — порядок сохраняется.
dataclass
¶
Hypothesis(hypothesis_id: str = '', statement: str = '', strategy: str = 'associative', support_score: float = 0.0, risk_score: float = 0.0, final_score: float = 0.0, evidence_ids: List[str] = list(), confidence: float = 0.0, metadata: Dict[str, Any] = dict())
Bases: ContractMixin
Гипотеза, сгенерированная на основе доказательств.
hypothesis_id: уникальный ID statement: текст гипотезы (человекочитаемый) strategy: стратегия генерации ("associative" | "deductive") support_score: сумма поддержки от доказательств [0..∞) risk_score: сумма рисков / противоречий [0..∞) final_score: support_score - risk_score (может быть отрицательным) evidence_ids: ID доказательств, на которых основана confidence: нормализованная уверенность [0..1] metadata: дополнительные данные
RetrievalAdapter¶
Структурированный retrieval: BM25 + векторный поиск.
Facade для retrieval: делегирует backend, добавляет metadata.
Гарантирует
- Все 11 канонических полей EvidencePack заполнены
- metadata содержит retrieval_backend и retrieved_at
- Результат отсортирован по relevance_score (desc)
Supports
- keyword-only search via retrieve(query)
- hybrid search via retrieve(query, query_vector=...) when backend is HybridRetrievalBackend
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
backend
|
RetrievalBackend
|
реализация RetrievalBackend |
required |
memory_manager
|
Any
|
опциональная ссылка на MemoryManager (для совместимости) |
None
|
retrieve(query: str, top_n: int = 10, query_vector: Optional[List[float]] = None) -> List[EvidencePack]
Извлечь evidence через backend с metadata enrichment.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
текстовый запрос |
required |
top_n
|
int
|
максимальное количество результатов |
10
|
query_vector
|
Optional[List[float]]
|
вектор запроса для гибридного поиска (optional) |
None
|
Returns:
| Type | Description |
|---|---|
List[EvidencePack]
|
Список EvidencePack с гарантированными каноническими полями |
HybridRetrievalBackend(keyword_backend: KeywordRetrievalBackend, vector_backend: VectorRetrievalBackend)
Hybrid retrieval backend: combines keyword search and vector search.
Strategy
- Run keyword search via KeywordRetrievalBackend → keyword_results
- Run vector search via VectorRetrievalBackend → vector_results
- Merge results using reciprocal rank fusion (RRF)
- Return top_n merged results
If vector_backend has no entries or query_vector is None, falls back to keyword-only search.
Usage
keyword_backend = KeywordRetrievalBackend(memory_manager) vector_backend = VectorRetrievalBackend() hybrid = HybridRetrievalBackend(keyword_backend, vector_backend) results = hybrid.search("нейрон", top_n=10) results = hybrid.search_hybrid("нейрон", query_vector=[...], top_n=10)
property
¶
Access to the vector backend for adding entries.
Keyword-only search (satisfies RetrievalBackend Protocol). For hybrid search with vectors, use search_hybrid().
search_hybrid(query: str, query_vector: Optional[List[float]] = None, top_n: int = 10) -> List[EvidencePack]
Hybrid search: keyword + vector with reciprocal rank fusion.
If query_vector is None or vector_backend is empty, falls back to keyword-only search.
BM25 (Okapi BM25) scorer для reranking кандидатов.
Использование
scorer = BM25Scorer(k1=1.5, b=0.75) scorer.fit(["документ один", "документ два", ...]) score = scorer.score("запрос", "документ один") scores = scorer.score_batch("запрос", ["документ один", "документ два"])
Если fit() не вызван — fallback на нормализованный keyword overlap. Лемматизация через pymorphy3 (optional, graceful fallback на lower+split).
Параметры BM25
k1: контроль насыщения TF (default 1.5, range [1.2, 2.0]) b: контроль нормализации длины документа (default 0.75, range [0, 1])
Построить IDF индекс на корпусе документов.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
documents
|
List[str]
|
список текстов документов |
required |
Returns:
| Type | Description |
|---|---|
'BM25Scorer'
|
self (для chaining) |
Вычислить BM25 score для одного документа.
Если fit() не вызван — fallback на keyword overlap.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
текстовый запрос |
required |
document
|
str
|
текст документа |
required |
Returns:
| Type | Description |
|---|---|
float
|
BM25 score (float >= 0.0) |
Вычислить BM25 scores для списка документов.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
текстовый запрос |
required |
documents
|
List[str]
|
список текстов документов |
required |
Returns:
| Type | Description |
|---|---|
List[float]
|
Список BM25 scores (same order as documents) |
ContradictionDetector¶
Обнаружение противоречий в базе знаний.
ContradictionDetector(numeric_threshold: float = 0.2, confidence_gap_threshold: float = 0.5, max_numbers_per_evidence: int = 2)
Детектор противоречий между доказательствами.
Проверяет все пары evidence с общим subject (concept_refs overlap). Три типа проверок: 1. Negation — одно evidence отрицает другое 2. Numeric — числовые значения расходятся >20% 3. Confidence gap — разница confidence >0.5
flag_evidence() — copy-on-write: возвращает новые EvidencePack.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
numeric_threshold
|
float
|
порог расхождения чисел (20%) |
0.2
|
confidence_gap_threshold
|
float
|
порог разницы confidence (0.5) |
0.5
|
max_numbers_per_evidence
|
int
|
пропустить numeric check если >N чисел |
2
|
Обнаружить противоречия между всеми парами evidence.
Проверяет только пары с общим subject (concept_refs overlap).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
evidence
|
List[EvidencePack]
|
список EvidencePack |
required |
Returns:
| Type | Description |
|---|---|
List[Contradiction]
|
Список обнаруженных Contradiction |
flag_evidence(evidence: List[EvidencePack], contradictions: List[Contradiction]) -> List[EvidencePack]
Пометить evidence, участвующие в противоречиях.
COPY-ON-WRITE: возвращает новые EvidencePack, не мутирует оригиналы. Добавляет contradiction type в contradiction_flags.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
evidence
|
List[EvidencePack]
|
оригинальный список EvidencePack |
required |
contradictions
|
List[Contradiction]
|
обнаруженные противоречия |
required |
Returns:
| Type | Description |
|---|---|
List[EvidencePack]
|
Новый список EvidencePack (копии с обновлёнными flags) |
dataclass
¶
Contradiction(evidence_a_id: str = '', evidence_b_id: str = '', type: str = '', severity: float = 0.0, description: str = '', shared_subject: str = '')
Bases: ContractMixin
Описание противоречия между двумя evidence.
evidence_a_id: ID первого evidence evidence_b_id: ID второго evidence type: тип противоречия ("negation" | "numeric" | "confidence_gap") severity: серьёзность [0.0, 1.0] description: человекочитаемое описание shared_subject: общий subject (concept из concept_refs overlap)
UncertaintyMonitor¶
Мониторинг тренда уверенности.
UncertaintyMonitor(window_size: int = 5, stagnation_threshold: float = 0.02, escalation_count: int = 3)
Мониторинг тренда confidence в reasoning loop.
Отслеживает историю confidence и определяет
- trend: rising / falling / stable / unknown
- should_early_stop: stagnation (delta < threshold) >= window_size
- should_escalate: falling trend >= escalation_count итераций подряд
Использование
monitor = UncertaintyMonitor() monitor.reset() # перед каждым reasoning run snapshot = monitor.update(state) # на каждой итерации
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
window_size
|
int
|
окно для определения stagnation |
5
|
stagnation_threshold
|
float
|
порог delta для "stable" (|delta| < threshold) |
0.02
|
escalation_count
|
int
|
сколько falling итераций подряд → escalate |
3
|
Обновить монитор с текущим состоянием reasoning loop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state
|
ReasoningState
|
текущий ReasoningState (используется current_confidence) |
required |
Returns:
| Type | Description |
|---|---|
UncertaintySnapshot
|
UncertaintySnapshot с текущим анализом |
Получить текущий тренд на основе последних данных.
Returns:
| Type | Description |
|---|---|
UncertaintyTrend
|
UncertaintyTrend enum value |
Рекомендация остановить reasoning из-за stagnation.
True если confidence стагнирует >= window_size итераций подряд.
Рекомендация эскалировать из-за falling trend.
True если confidence падает >= escalation_count итераций подряд.
dataclass
¶
UncertaintySnapshot(confidence: float = 0.0, trend: str = 'unknown', delta: float = 0.0, iteration: int = 0, should_stop: bool = False, should_escalate: bool = False, history_length: int = 0)
Bases: ContractMixin
Снимок состояния неопределённости на текущей итерации.
confidence: текущая confidence (каноническая величина) trend: тренд как str ("rising"/"falling"/"stable"/"unknown") delta: изменение confidence с предыдущей итерации iteration: номер итерации should_stop: рекомендация остановить reasoning (stagnation) should_escalate: рекомендация эскалировать (falling trend) history_length: длина истории confidence
CognitiveContext¶
Контекст когнитивного цикла: состояние, ограничения, исходы.
dataclass
¶
CognitiveContext(session_id: str = '', cycle_id: str = '', trace_id: str = '', active_goal: Optional[Any] = None, goal_chain: List[Any] = list())
Bases: ContractMixin
Состояние текущего цикла мышления. Передаётся всем компонентам когнитивного ядра.
active_goal и goal_chain типизированы как Any, чтобы избежать циклического импорта с goal_manager.py (Goal определён там).
Bases: str, Enum
Коды завершения reasoning loop. Используются в check_stop_conditions(), replan() и логах.
Разделены на нормальные завершения и ошибки/сбои.
dataclass
¶
PolicyConstraints(min_confidence: float = 0.4, hedge_threshold: float = 0.6, max_retries: int = 2, goal_limits: GoalTypeLimits = (lambda: GoalTypeLimits())())
Bases: ContractMixin
Ограничения поведения для текущего цикла.
Атрибуты
min_confidence: минимальная уверенность для прямого ответа (ActionSelector). hedge_threshold: порог ниже которого ResponseValidator добавляет оговорку. max_retries: максимум повторов replan(). goal_limits: stop conditions для текущего типа цели.
── Карта порогов уверенности системы ──────────────────────────────────
Централизованные (живут здесь, в PolicyConstraints + GoalTypeLimits):
GoalTypeLimits.confidence_threshold (0.70–0.80) Когда остановить reasoning loop. Разный для каждого типа цели: answer_question=0.75, verify_claim=0.80, explore_topic=0.70, learn_fact=0.70. Используется в planner.py.
PolicyConstraints.min_confidence (0.4) Порог для ActionSelector: >= min_confidence → RESPOND_DIRECT, >= min_confidence * 0.6 → RESPOND_HEDGED, ниже → ASK/REFUSE.
PolicyConstraints.hedge_threshold (0.6) Порог для ResponseValidator: если confidence < hedge_threshold и ответ не содержит маркеров оговорки — добавить "Возможно, ". Должен быть >= min_confidence для согласованности.
Доменно-специфичные (живут в своих модулях, не централизованы):
ContradictionDetector.confidence_gap_threshold (0.50) Другой домен: сравнение пар доказательств между собой. Не связан с порогами принятия решений. → brain/cognition/contradiction_detector.py
OutputTraceBuilder._compute_uncertainty_level (0.85/0.65/0.45/0.25) Метки для отображения уровня неопределённости (display-only). Не влияют на логику принятия решений. → brain/output/trace_builder.py
DialogueResponder.HEDGING_PHRASES (0.75/0.60/0.45/0.30) Шаблоны фраз-оговорок по диапазонам confidence. Применяются только к уже принятому решению RESPOND_HEDGED. → brain/output/dialogue_responder.py
───────────────────────────────────────────────────────────────────────