src/graph/engine.py - master - Ilmer/tandemBotLLM

src/graph/engine.py
143 строки7 KB
vkuzin
base proect
14 май 2026, 07:36
14 май 2026, 07:360893926
from __future__ import annotations

import logging
from typing import Any

from graph.types import GraphRunContext
from services.escalation_service import EscalationService
from tools.registry import ToolInvocation

logger = logging.getLogger(__name__)

MAX_STEPS = 32


class GraphEngine:
    """Deterministic scenario interpreter (variant D)."""

    def __init__(self, scenario: dict[str, Any], *, escalation: EscalationService) -> None:
        self._scenario = scenario
        self._nodes: dict[str, dict[str, Any]] = scenario["nodes"]
        self._escalation = escalation

    @property
    def meta(self) -> dict[str, Any]:
        return self._scenario.get("meta", {})

    async def run_turn(self, ctx: GraphRunContext) -> None:
        node_id = ctx.session.current_node
        steps = 0
        text_in = (ctx.incoming_text or "").strip()
        consumed_user_line = False

        while steps < MAX_STEPS:
            steps += 1
            node = self._nodes.get(node_id)
            if not node:
                ctx.outbound.append(
                    "Не могу выполнить запрос в текущих условиях: сценарий повреждён (неизвестный узел). "
                    "Напишите «оператор»."
                )
                ctx.session.current_node = "wait_user"
                return

            ntype = node["type"]
            if ntype == "send_text":
                ctx.outbound.append(str(node.get("text", "")).strip())
                node_id = str(node["next"])
            elif ntype == "send_text_template":
                excerpt = ctx.kb_excerpt or "(пусто)"
                tpl = str(node.get("template", ""))
                ctx.outbound.append(tpl.format(excerpt=excerpt).strip())
                node_id = str(node["next"])
            elif ntype == "wait_input":
                if not text_in:
                    ctx.session.current_node = node_id
                    return
                if consumed_user_line:
                    ctx.session.current_node = node_id
                    return
                consumed_user_line = True
                text_in = ""
                node_id = str(node["next"])
            elif ntype == "branch_operator_or_kb":
                low = ctx.incoming_text.strip().lower()
                kws = [str(k).lower() for k in node.get("operator_keywords", [])]
                if any(k in low for k in kws):
                    node_id = str(node["next_operator"])
                else:
                    node_id = str(node["next_default"])
            elif ntype == "kb_search":
                limit = int(node.get("limit", 5))
                q = (ctx.incoming_text or "").strip()
                hits = await ctx.services.knowledge.search(q, limit=limit)
                if hits:
                    first = hits[0]
                    ctx.kb_excerpt = str(first.get("excerpt") or first.get("body", ""))[:2000]
                    node_id = str(node["next_hit"])
                else:
                    ctx.kb_excerpt = None
                    node_id = str(node["next_miss"])
            elif ntype == "escalate":
                tail = (ctx.session.history or [])[-5:]
                await self._escalation.open_ticket(
                    session_id=ctx.session.id,
                    person_id=ctx.session.person_id,
                    reason="graph_escalate",
                    context={
                        "incoming": ctx.incoming_text.strip(),
                        "tail": tail,
                        "messenger": ctx.session.messenger.value,
                        "language": ctx.session.slots.get("language"),
                    },
                )
                ctx.outbound.append(str(node.get("user_message", "")).strip())
                node_id = str(node["next"])
            elif ntype == "tool_call":
                reg = ctx.services.tools
                name = str(node.get("tool", "")).strip()
                args = node.get("arguments") if isinstance(node.get("arguments"), dict) else {}
                nxt_ok = str(node["next"])
                nxt_err = str(node.get("next_on_error", nxt_ok))
                if not reg or not name:
                    ctx.outbound.append(
                        "Не могу выполнить запрос в текущих условиях: инструмент недоступен. "
                        "Напишите «оператор»."
                    )
                    node_id = nxt_err
                else:
                    try:
                        await reg.invoke(ToolInvocation(name=name, arguments=args))
                        node_id = nxt_ok
                    except Exception as exc:  # noqa: BLE001 — сценарий, не раскрываем стек пользователю
                        logger.warning("tool_call %s failed: %s", name, exc)
                        ctx.outbound.append(
                            "Не могу выполнить запрос в текущих условиях: ошибка инструмента. "
                            "Переформулируйте или напишите «оператор»."
                        )
                        node_id = nxt_err
            elif ntype.startswith("llm_") or ntype.startswith("assistant_"):
                if not ctx.services.settings_llm_enabled:
                    ctx.outbound.append(
                        "Не могу выполнить запрос в текущих условиях: узел с LLM отключён конфигурацией. "
                        "Переформулируйте или напишите «оператор»."
                    )
                    node_id = str(node.get("next_on_disabled", "wait_user"))
                else:
                    raise NotImplementedError(
                        "LLM nodes are optional and not implemented in this skeleton."
                    )
            else:
                logger.error("Unknown node type %s in %s", ntype, node_id)
                ctx.outbound.append(
                    "Не могу выполнить запрос в текущих условиях: неподдерживаемый тип узла. "
                    "Напишите «оператор»."
                )
                node_id = "wait_user"

            ctx.session.current_node = node_id

        ctx.outbound.append(
            "Не могу выполнить запрос в текущих условиях: превышен лимит шагов сценария. Напишите «оператор»."
        )
        ctx.session.current_node = "wait_user"