core/engine/src/runtime/node_executor.py - main - alexefan136/flowstack

core/engine/src/runtime/node_executor.py
272 строки11 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""Flow Node Executor — выполняет узлы flow, делегируя AgentRuntime/LLM/tools.

Связывает GraphExecutor (обход графа) с конкретными исполнителями:
- AGENT → AgentRuntime (LLM loop + function calling)
- LLM → прямой вызов LLMClient
- TOOL → tool_executor
- TRANSFORM → safe_eval / mapping
- SUBFLOW → вложенный GraphExecutor
"""

from __future__ import annotations

import re
from typing import Any

import structlog

from src.llm.client import LLMClient, get_llm_client
from src.llm.config import LLMConfig
from src.primitives import (
    AgentContext,
    FlowNode,
    TaskStatus,
    safe_eval,
)
from src.runtime.agent_runtime import AgentConfig, AgentRuntime, ToolExecutor
from src.runtime.graph_executor import GraphExecutor

logger = structlog.get_logger()

# Паттерн для подстановки переменных {{var_name}}
_VAR_PATTERN = re.compile(r"\{\{(\w+)\}\}")


class FlowNodeExecutor:
    """
    Исполнитель узлов flow.

    Реализует сигнатуру ``NodeExecutor``: ``(node, context) -> dict``.
    Выбирает стратегию выполнения по типу узла.

    Args:
        agent_runtime: для AGENT-узлов (создаётся по умолчанию)
        llm_client: для LLM-узлов
        tool_executor: для TOOL-узлов и function calling в агентах
    """

    def __init__(
        self,
        agent_runtime: AgentRuntime | None = None,
        llm_client: LLMClient | None = None,
        tool_executor: ToolExecutor | None = None,
    ) -> None:
        self.tool_executor = tool_executor
        self.llm_client = llm_client or get_llm_client()
        self.agent_runtime = agent_runtime or AgentRuntime(
            llm_client=self.llm_client, tool_executor=tool_executor
        )
        logger.info("flow_node_executor.initialized")

    async def __call__(self, node: FlowNode, context: dict[str, Any]) -> dict[str, Any]:
        """Выполнить узел (NodeExecutor interface)."""
        # Dispatch по типу узла: AGENT → _execute_agent, LLM → _execute_llm, ...
        # Для неизвестных типов — _execute_unsupported (default)
        handler = getattr(self, f"_execute_{node.type.value}", self._execute_unsupported)
        logger.debug("flow_node_executor.execute", node_id=node.id, node_type=node.type.value)
        return await handler(node, context)

    # ========================================================================
    # Template rendering
    # ========================================================================

    def _render(self, template: str, context: dict[str, Any]) -> str:
        """Подставить переменные {{var}} из context в шаблон."""
        if not template:
            return ""

        def replace_var(match: re.Match[str]) -> str:
            var_name = match.group(1)
            value = context.get(var_name)
            if value is None:
                return match.group(0)  # оставляем {{var}} если нет значения
            return str(value)

        return _VAR_PATTERN.sub(replace_var, template)

    def _render_dict(self, data: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
        """Отрендерить строковые значения в dict."""
        return {k: self._render(v, context) if isinstance(v, str) else v for k, v in data.items()}

    # ========================================================================
    # AGENT node
    # ========================================================================

    async def _execute_agent(self, node: FlowNode, context: dict[str, Any]) -> dict[str, Any]:
        """Выполнить AGENT-узел через AgentRuntime."""
        config = node.config or {}

        agent_config = AgentConfig(
            agent_id=config.get("agent_id", node.id),
            agent_name=config.get("agent_name", node.label or node.id),
            system_prompt=self._render(config.get("system_prompt", ""), context),
            model=config.get("model", "deepseek-ai/DeepSeek-V4-Pro"),
            tools=config.get("tools", []),
            temperature=config.get("temperature", 0.7),
            max_tokens=config.get("max_tokens", 4096),
            max_iterations=config.get("max_iterations", 10),
        )

        # Входное сообщение: prompt-шаблон или fallback на input/query из context
        prompt = self._render(config.get("prompt", config.get("input_template", "")), context)
        if not prompt:
            prompt = str(context.get("input", context.get("query", "")))

        messages = [{"role": "user", "content": prompt}]

        agent_context = AgentContext(
            agent_id=agent_config.agent_id,
            workspace_id=str(context.get("workspace_id", "default")),
        )

        state = await self.agent_runtime.run(agent_config, messages, agent_context)

        if state.status != TaskStatus.COMPLETED:
            raise RuntimeError(
                f"Agent '{agent_config.agent_id}' failed: {state.error or 'unknown error'}"
            )

        return {
            "output": state.get_output("output", ""),
            "reasoning": state.get_output("reasoning_content"),
            "agent_id": agent_config.agent_id,
            "tokens": state.tokens_used,
        }

    # ========================================================================
    # LLM node
    # ========================================================================

    async def _execute_llm(self, node: FlowNode, context: dict[str, Any]) -> dict[str, Any]:
        """Выполнить LLM-узел (прямой вызов, без agent loop)."""
        config = node.config or {}

        prompt = self._render(config.get("prompt", ""), context)
        system_prompt = self._render(config.get("system_prompt", ""), context)

        messages: list[dict[str, Any]] = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        llm_config = LLMConfig(
            model=config.get("model", "deepseek-ai/DeepSeek-V4-Pro"),
            max_tokens=config.get("max_tokens", 4096),
            temperature=config.get("temperature", 0.7),
            top_p=config.get("top_p", 0.9),
            timeout=300.0,
        )

        response = await self.llm_client.chat_completion(messages, config=llm_config)

        return {
            "output": response.get("content", ""),
            "reasoning": response.get("reasoning_content"),
            "tokens": response.get("tokens_total", 0),
        }

    # ========================================================================
    # TOOL node
    # ========================================================================

    async def _execute_tool(self, node: FlowNode, context: dict[str, Any]) -> dict[str, Any]:
        """Выполнить TOOL-узел через tool_executor."""
        config = node.config or {}
        tool_name = config.get("tool_name", config.get("tool", ""))

        if not tool_name:
            raise ValueError(f"Tool node '{node.id}' has no 'tool_name' in config")
        if self.tool_executor is None:
            raise ValueError(f"No tool executor configured for tool '{tool_name}'")

        arguments = self._render_dict(config.get("arguments", {}), context)
        result = await self.tool_executor(tool_name, arguments)

        return {"output": result, "tool": tool_name}

    # ========================================================================
    # TRANSFORM node
    # ========================================================================

    async def _execute_transform(self, node: FlowNode, context: dict[str, Any]) -> dict[str, Any]:
        """
        Выполнить TRANSFORM-узел.

        Два режима:
        - ``expression``: безопасное вычисление через safe_eval
        - ``mapping``: рендеринг dict-шаблона {{var}}
        """
        config = node.config or {}

        # Режим 1: выражение (safe_eval)
        expression = config.get("expression")
        if expression:
            result = safe_eval(expression, context)
            return {"output": result}

        # Режим 2: mapping
        mapping = config.get("mapping", {})
        if mapping:
            return self._render_dict(mapping, context)

        # Passthrough: весь context
        return {"output": context}

    # ========================================================================
    # SUBFLOW node
    # ========================================================================

    async def _execute_subflow(self, node: FlowNode, context: dict[str, Any]) -> dict[str, Any]:
        """Выполнить SUBFLOW-узел (вложенный flow через GraphExecutor)."""
        config = node.config or {}
        subflow = config.get("flow")

        if subflow is None:
            raise ValueError(f"Subflow node '{node.id}' has no 'flow' in config")

        # Вложенный executor с тем же node_executor (рекурсия)
        nested_executor = GraphExecutor(node_executor=self)
        task = await nested_executor.run(subflow, context)

        if task.status != TaskStatus.COMPLETED:
            raise RuntimeError(f"Subflow '{node.id}' failed: {task.error or 'unknown error'}")

        return {"output": task.output, "subflow_id": node.id}

    # ========================================================================
    # Unsupported
    # ========================================================================

    async def _execute_unsupported(self, node: FlowNode, context: dict[str, Any]) -> dict[str, Any]:
        """Обработчик для неподдерживаемых типов узлов."""
        raise ValueError(f"Unsupported node type: {node.type.value} (node '{node.id}')")


# ============================================================================
# Convenience factory
# ============================================================================


def create_graph_executor(
    agent_runtime: AgentRuntime | None = None,
    llm_client: LLMClient | None = None,
    tool_executor: ToolExecutor | None = None,
) -> GraphExecutor:
    """
    Создать GraphExecutor с полным FlowNodeExecutor.

    Связывает обход графа с исполнителями узлов (AgentRuntime/LLM/tools).

    Usage:
        executor = create_graph_executor(tool_executor=my_tools)
        task = await executor.run(flow, {"query": "..."})
    """
    node_executor = FlowNodeExecutor(
        agent_runtime=agent_runtime,
        llm_client=llm_client,
        tool_executor=tool_executor,
    )
    return GraphExecutor(node_executor=node_executor)


__all__ = ["FlowNodeExecutor", "create_graph_executor"]