Справочник по рабочим процессам

classify_intent -> pre_retrieval -> route_by_intent -> scenarios -> END

graph TD
    A[Запрос пользователя] --> B[classify_intent]
    B --> C[pre_retrieval]
    C --> D[route_by_intent]
    D --> S01[onboarding_workflow]
    D --> S02[security_workflow]
    D --> S03[documentation_workflow]
    D --> S04[feature_dev_workflow]
    D --> S05[refactoring_workflow]
    D --> S06[performance_workflow]
    D --> S07[test_coverage_workflow]
    D --> S08[compliance_workflow]
    D --> S09[code_review_workflow]
    D --> S10[cross_repo_workflow]
    D --> S11[architecture_workflow]
    D --> S12[tech_debt_workflow]
    D --> S13[mass_refactoring_workflow]
    D --> S14[security_incident_workflow]
    D --> S15[debugging_workflow]
    D --> S16[entry_points_workflow]
    D --> S17[file_editing_workflow]
    D --> S18[optimization_workflow]
    D --> S19[standards_check_workflow]
    D --> S20[dependencies_workflow]
    D --> S21[interface_docs_sync_workflow]
    S01 --> E[END]
    S02 --> E
    S03 --> E
    S04 --> E
    S05 --> E
    S06 --> E
    S07 --> E
    S08 --> E
    S09 --> E
    S10 --> E
    S11 --> E
    S12 --> E
    S13 --> E
    S14 --> E
    S15 --> E
    S16 --> E
    S17 --> E
    S18 --> E
    S19 --> E
    S20 --> E
    S21 --> E

class MultiScenarioCopilot:
    def __init__(self):
        self.graph = build_multi_scenario_graph()

    def run(self, query: str, context: Optional[Dict] = None) -> Dict[str, Any]:
        ...

from src.workflow import MultiScenarioCopilot

copilot = MultiScenarioCopilot()

# Автоматическое определение сценария по запросу
result = copilot.run("Найти SQL-инъекции")

# Принудительный выбор сценария
result = copilot.run(
    "Проанализировать модуль",
    context={"scenario_id": "scenario_2"}
)

# Структура результата
{
    "query": "Найти SQL-инъекции",
    "intent": "security_audit",
    "scenario_id": "scenario_2",
    "confidence": 0.92,
    "answer": "Обнаружено 3 потенциальных SQL-инъекции...",
    "evidence": ["Функция exec_simple_query в строке 142..."],
    "metadata": {...}
}

class MultiScenarioState(TypedDict):
    # Входные данные
    query: str
    context: Optional[Dict[str, Any]]
    language: Optional[str]               # "en" или "ru"

    # Классификация намерений
    intent: Optional[str]
    scenario_id: Optional[str]
    confidence: Optional[float]
    classification_method: Optional[str]

    # Данные CPG
    cpg_results: Optional[List[Dict]]
    subsystems: Optional[List[str]]
    methods: Optional[List[Dict]]
    call_graph: Optional[Any]

    # Выходные данные
    answer: Optional[str]
    evidence: Optional[List[str]]
    metadata: Optional[Dict[str, Any]]
    retrieved_functions: Optional[List[str]]

    # Обработка ошибок
    error: Optional[str]
    retry_count: int

    # Конфигурация
    enrichment_config: Optional[Dict[str, Any]]
    vector_store: Optional[Any]

    # Мультитенантная изоляция
    db_path: Optional[str]
    collection_prefix: Optional[str]

    # Результаты предварительного поиска (Phase E)
    pre_retrieval_results: Optional[List[Dict[str, Any]]]

from src.workflow.state import create_initial_state

state = create_initial_state(
    query="Найти утечки памяти",
    context={"subsystem": "executor"}
)

from src.workflow.orchestration.intent_classifier import classify_intent_node

# Вызывается внутренне графом
state = classify_intent_node(state)
# Заполняет: state["intent"], state["scenario_id"], state["confidence"],
#             state["classification_method"]

from src.workflow.orchestration.router import route_by_intent

# Возвращает имя узла сценария
next_node = route_by_intent(state)
# Например: "security_workflow", "onboarding_workflow"

src/workflow/scenarios/
+-- _base/                      # Базовый класс обработчика
|   +-- handler.py              # BaseHandler, HandlerResult
+-- _intent/                    # Классификация намерений
+-- onboarding/                 # S01 Онбординг
+-- security/                   # S02 Безопасность
+-- documentation_handlers/     # S03 Документация
+-- feature_dev_handlers/       # S04 Разработка функций
+-- refactoring_handlers/       # S05 Рефакторинг
+-- performance_handlers/       # S06 Производительность
+-- coverage_handlers/          # S07 Покрытие тестами
+-- compliance_handlers/        # S08 Соответствие
+-- code_review_handlers/       # S09 Проверка кода
+-- cross_repo_handlers/        # S10 Кросс-репозиторный анализ
+-- architecture_handlers/      # S11 Архитектура
+-- tech_debt_handlers/         # S12 Техдолг
+-- mass_refactoring.py         # S13 Рефакторинг корпоративного масштаба
+-- security_incident.py        # S14 Реагирование на инциденты
+-- debugging_handlers/         # S15 Отладка
+-- entry_points.py             # S16 Точки входа
+-- file_editing.py             # S17 Редактирование файлов
+-- code_optimization.py        # S18 Оптимизация кода
+-- standards_check.py          # S19 Проверка стандартов
+-- dependencies_analysis.py    # S20 Анализ зависимостей
+-- interface_docs_sync_composite.py  # S21 Синхронизация документации
+-- audit_composite.py          # Аудит (композитный)
+-- story_validation_composite.py  # Валидация историй (композитный)

class BaseHandler(ABC, Generic[T]):
    def __init__(self, cpg: Any, state: MultiScenarioState):
        ...

@dataclass
class HandlerResult(Generic[T]):
    data: Optional[T] = None
    cpg_results: List[Dict[str, Any]] = field(default_factory=list)
    retrieved_functions: List[str] = field(default_factory=list)
    answer: str = ""
    evidence: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    should_return: bool = True
    llm_context: Dict[str, Any] = field(default_factory=dict)

from src.workflow.scenarios._base.handler import BaseHandler, HandlerResult

class MyHandler(BaseHandler):
    def handle(self, query_info: Dict[str, Any]) -> HandlerResult:
        results = self.cpg.get_methods_by_subsystem("executor")
        return HandlerResult(
            answer="Найдены методы...",
            evidence=["Метод heap_insert в строке 142"],
            metadata={"handler": "my_handler"}
        )

python -m src.cli audit --db PATH [--language ru] [--format json]

python -m src.cli docs-sync --db PATH [--check] [--format json] [--interfaces rest_api,cli]

python -m src.cli.import_commands dogfood validate-stories

python -m src.cli exec --prompt "Review security" --base-ref origin/main \
    --sarif-file out.sarif --comment-file comment.md --sandbox read-only

# Встроено в граф -- настраивается через state["retry_count"]
# По умолчанию: до 2 повторных попыток с адаптивным уточнением запроса

# Автоматическая цепочка запасных вариантов:
# 1. SQL-запрос, сгенерированный LLM
# 2. Запрос по шаблону из примеров
# 3. Прямой вызов метода CPG

from langgraph.graph import StateGraph
from src.workflow.state import MultiScenarioState

def create_custom_workflow():
    workflow = StateGraph(MultiScenarioState)

    workflow.add_node("analyze", my_analyze_node)
    workflow.add_node("process", my_process_node)
    workflow.add_node("interpret", my_interpret_node)

    workflow.add_edge("analyze", "process")
    workflow.add_edge("process", "interpret")

    workflow.set_entry_point("analyze")
    workflow.set_finish_point("interpret")

    return workflow.compile()

result = create_custom_workflow().invoke({"query": "..."})

def my_route(state: MultiScenarioState) -> str:
    if state["intent"] == "security_audit":
        return "security_node"
    elif state["intent"] == "performance":
        return "performance_node"
    else:
        return "general_node"

workflow.add_conditional_edges(
    "analyze",
    my_route,
    {
        "security_node": "security",
        "performance_node": "performance",
        "general_node": "general"
    }
)