/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/flows/create.py
1 182 строки
42 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
# core/engine/src/flows/create.py """Flow Creator — удобное создание custom flows через шаблоны и CLI. Предоставляет набор готовых шаблонов (templates) для типовых сценариев и CLI утилиту для быстрого создания custom flows без написания кода. Основные возможности: - Готовые шаблоны для типовых задач (research, analysis, review, etc.) - CLI для интерактивного создания flow - Генерация YAML конфигов из шаблонов - Фабрика для программного создания из шаблонов - Кастомизация шаблонов под конкретные нужды - Валидация и автоматическая регистрация Примеры использования: 1. Программное создание из шаблона: from src.flows.create import create_from_template, list_templates # Список доступных шаблонов templates = list_templates() # Создать flow из шаблона flow = create_from_template( template_name="research_deep", flow_id="my_research", customizations={ "focus": "medical", "language": "ru", } ) 2. CLI создание: # Интерактивное создание python -m src.flows.create # Из шаблона python -m src.flows.create --template research_deep --id my_research # Генерация YAML без регистрации python -m src.flows.create --template research_deep --yaml-only -o flow.yaml 3. Кастомизация шаблона: flow = create_from_template( template_name="code_review", flow_id="security_review", customizations={ "focus": "security", "extra_agents": ["judge"], } ) """ from __future__ import annotations import argparse import asyncio import json import sys from typing import Any, Callable import structlog from src.flows.custom import ( CustomFlowBuilder, FlowValidationError, create_custom_flow, export_flow_to_yaml, register_custom_flow, ) from src.primitives.flow import Flow logger = structlog.get_logger() # ============================================================================ # Templates Registry # ============================================================================ class FlowTemplate: """Шаблон для создания flow.""" def __init__( self, name: str, display_name: str, description: str, category: str, agents: list[dict[str, Any]], tags: list[str] | None = None, customizations: dict[str, Any] | None = None, ): """ Args: name: Уникальный ID шаблона (snake_case) display_name: Человекочитаемое имя description: Описание шаблона category: Категория (research, review, analysis, generation, etc.) agents: Список конфигураций агентов tags: Теги для поиска customizations: Доступные параметры кастомизации """ self.name = name self.display_name = display_name self.description = description self.category = category self.agents = agents self.tags = tags or [] self.customizations = customizations or {} def to_dict(self) -> dict[str, Any]: """Сериализация шаблона.""" return { "name": self.name, "display_name": self.display_name, "description": self.description, "category": self.category, "agents": self.agents, "tags": self.tags, "customizations": self.customizations, } # Предопределённые шаблоны TEMPLATES: dict[str, FlowTemplate] = {} def _register_template(template: FlowTemplate) -> None: """Зарегистрировать шаблон.""" TEMPLATES[template.name] = template # ============================================================================ # Research Templates # ============================================================================ _register_template(FlowTemplate( name="research_basic", display_name="Базовое исследование", description="Простое исследование: планирование → сбор информации → отчёт", category="research", tags=["research", "basic", "fast"], agents=[ {"type": "planner", "name": "Research Planner"}, {"type": "researcher", "name": "Information Gatherer"}, {"type": "writer", "name": "Report Writer"}, ], customizations={ "language": { "description": "Язык ответа (ru, en)", "default": "ru", "type": "string", }, "depth": { "description": "Глубина исследования (quick, standard, deep)", "default": "standard", "type": "enum", "options": ["quick", "standard", "deep"], }, }, )) _register_template(FlowTemplate( name="research_deep", display_name="Глубокое исследование", description="Полный цикл: планирование → сбор → анализ → отчёт → проверка", category="research", tags=["research", "deep", "comprehensive"], agents=[ {"type": "planner", "name": "Research Planner"}, {"type": "researcher", "name": "Information Gatherer"}, {"type": "analyst", "name": "Data Analyst"}, {"type": "writer", "name": "Report Writer"}, {"type": "reviewer", "name": "Quality Reviewer"}, ], customizations={ "language": {"default": "ru", "type": "string"}, "focus": { "description": "Фокус исследования", "default": "general", "type": "string", }, }, )) _register_template(FlowTemplate( name="research_comparative", display_name="Сравнительное исследование", description="Сравнение нескольких подходов/технологий с экспертной оценкой", category="research", tags=["research", "comparison", "analysis"], agents=[ {"type": "planner", "name": "Comparison Planner"}, {"type": "researcher", "name": "Options Researcher"}, {"type": "analyst", "name": "Comparative Analyst"}, {"type": "debater", "name": "Options Debater"}, {"type": "judge", "name": "Final Judge"}, ], customizations={ "options_count": { "description": "Количество сравниваемых вариантов", "default": 3, "type": "integer", }, }, )) # ============================================================================ # Review Templates # ============================================================================ _register_template(FlowTemplate( name="content_review", display_name="Ревью контента", description="Критический обзор текста/статьи/документа", category="review", tags=["review", "content", "critique"], agents=[ {"type": "critic", "name": "Content Critic"}, {"type": "reviewer", "name": "Quality Reviewer"}, {"type": "writer", "name": "Report Writer"}, ], customizations={ "review_type": { "description": "Тип контента (article, document, email, etc.)", "default": "article", "type": "string", }, }, )) _register_template(FlowTemplate( name="code_review_security", display_name="Security Code Review", description="Специализированное ревью кода с фокусом на безопасности", category="review", tags=["review", "code", "security"], agents=[ { "type": "researcher", "name": "Code Scanner", "system_prompt": "Ты — security researcher. Изучи код и найди потенциальные уязвимости по OWASP Top 10." }, { "type": "analyst", "name": "Vulnerability Analyst", "system_prompt": "Проанализируй найденные security issues по CVSS scoring." }, { "type": "reviewer", "name": "Security Reviewer", "system_prompt": "Проведи security code review с приоритизацией P0/P1/P2." }, { "type": "writer", "name": "Security Report Writer", "system_prompt": "Сформируй финальный security audit report." }, ], customizations={ "standards": { "description": "Стандарты безопасности (OWASP, NIST, etc.)", "default": "OWASP", "type": "string", }, }, )) _register_template(FlowTemplate( name="code_review_performance", display_name="Performance Code Review", description="Ревью кода с фокусом на производительности", category="review", tags=["review", "code", "performance"], agents=[ { "type": "researcher", "name": "Performance Researcher", "system_prompt": "Изучи код и найди потенциальные узкие места производительности." }, { "type": "analyst", "name": "Performance Analyst", "system_prompt": "Проанализируй Big-O сложность, memory usage, bottlenecks." }, { "type": "critic", "name": "Optimization Critic", "system_prompt": "Предложи конкретные оптимизации с оценкой эффекта." }, { "type": "writer", "name": "Performance Report Writer", }, ], )) # ============================================================================ # Analysis Templates # ============================================================================ _register_template(FlowTemplate( name="data_analysis", display_name="Анализ данных", description="Глубокий анализ данных с выявлением паттернов и инсайтов", category="analysis", tags=["analysis", "data", "insights"], agents=[ {"type": "researcher", "name": "Data Explorer"}, {"type": "analyst", "name": "Pattern Analyst"}, {"type": "writer", "name": "Insights Writer"}, ], customizations={ "data_type": { "description": "Тип данных (quantitative, qualitative, mixed)", "default": "mixed", "type": "enum", "options": ["quantitative", "qualitative", "mixed"], }, }, )) _register_template(FlowTemplate( name="business_analysis", display_name="Бизнес-анализ", description="Анализ бизнес-кейса с SWOT, рисками и рекомендациями", category="analysis", tags=["analysis", "business", "strategy"], agents=[ { "type": "researcher", "name": "Market Researcher", "system_prompt": "Проведи исследование рынка и контекста бизнес-кейса." }, { "type": "analyst", "name": "Business Analyst", "system_prompt": "Проведи SWOT анализ, оцени риски и возможности." }, { "type": "critic", "name": "Strategy Critic", "system_prompt": "Критически оцени стратегию и предложи альтернативы." }, { "type": "writer", "name": "Business Report Writer", "system_prompt": "Сформируй бизнес-отчёт с рекомендациями." }, ], )) # ============================================================================ # Generation Templates # ============================================================================ _register_template(FlowTemplate( name="content_generation", display_name="Генерация контента", description="Создание контента с планированием, написанием и ревью", category="generation", tags=["generation", "content", "writing"], agents=[ {"type": "planner", "name": "Content Planner"}, {"type": "writer", "name": "Content Writer"}, {"type": "reviewer", "name": "Content Reviewer"}, ], customizations={ "content_type": { "description": "Тип контента (article, post, email, documentation)", "default": "article", "type": "enum", "options": ["article", "post", "email", "documentation"], }, "tone": { "description": "Тон (formal, casual, technical, friendly)", "default": "professional", "type": "string", }, }, )) _register_template(FlowTemplate( name="technical_documentation", display_name="Техническая документация", description="Создание технической документации с примерами кода", category="generation", tags=["generation", "documentation", "technical"], agents=[ { "type": "planner", "name": "Doc Planner", "system_prompt": "Спланируй структуру технической документации." }, { "type": "writer", "name": "Technical Writer", "system_prompt": "Напиши техническую документацию с примерами кода." }, { "type": "reviewer", "name": "Doc Reviewer", "system_prompt": "Проверь документацию на точность и понятность." }, ], )) # ============================================================================ # Decision Templates # ============================================================================ _register_template(FlowTemplate( name="decision_making", display_name="Принятие решения", description="Дебаты с аргументами за/против и финальным вердиктом", category="decision", tags=["decision", "debate", "analysis"], agents=[ {"type": "planner", "name": "Decision Planner"}, {"type": "debater", "name": "Pro Advocate"}, {"type": "debater", "name": "Con Advocate"}, {"type": "analyst", "name": "Arguments Analyst"}, {"type": "judge", "name": "Decision Judge"}, ], )) _register_template(FlowTemplate( name="risk_assessment", display_name="Оценка рисков", description="Комплексная оценка рисков с митигациями", category="decision", tags=["decision", "risk", "assessment"], agents=[ { "type": "researcher", "name": "Risk Researcher", "system_prompt": "Исследуй потенциальные риски в предоставленной ситуации." }, { "type": "analyst", "name": "Risk Analyst", "system_prompt": "Проведи количественную оценку рисков (вероятность × влияние)." }, { "type": "critic", "name": "Mitigation Critic", "system_prompt": "Предложи стратегии митигации для каждого риска." }, { "type": "judge", "name": "Risk Judge", "system_prompt": "Вынеси финальную оценку с приоритизированным планом действий." }, ], )) # ============================================================================ # Template Management Functions # ============================================================================ def list_templates(category: str | None = None) -> list[FlowTemplate]: """ Получить список доступных шаблонов. Args: category: Фильтр по категории (опционально) Returns: Список шаблонов, отсортированный по имени """ templates = list(TEMPLATES.values()) if category: templates = [t for t in templates if t.category == category] return sorted(templates, key=lambda t: (t.category, t.name)) def get_template(name: str) -> FlowTemplate | None: """ Получить шаблон по имени. Args: name: Имя шаблона Returns: FlowTemplate или None если не найден """ return TEMPLATES.get(name) def list_categories() -> list[str]: """ Получить список всех категорий шаблонов. Returns: Отсортированный список категорий """ return sorted(set(t.category for t in TEMPLATES.values())) def search_templates(query: str) -> list[FlowTemplate]: """ Поиск шаблонов по query. Ищет в имени, display_name, description и tags. Args: query: Поисковый запрос Returns: Список найденных шаблонов """ query_lower = query.lower() results = [] for template in TEMPLATES.values(): searchable = " ".join([ template.name, template.display_name, template.description, " ".join(template.tags), template.category, ]).lower() if query_lower in searchable: results.append(template) return results # ============================================================================ # Flow Creation from Template # ============================================================================ def create_from_template( template_name: str, flow_id: str, flow_name: str | None = None, description: str | None = None, customizations: dict[str, Any] | None = None, auto_register: bool = False, ) -> Flow: """ Создать flow из шаблона. Args: template_name: Имя шаблона flow_id: ID создаваемого flow flow_name: Имя flow (опционально, берётся из шаблона) description: Описание flow (опционально, берётся из шаблона) customizations: Параметры кастомизации auto_register: Автоматически зарегистрировать в registry Returns: Flow instance Raises: ValueError: Если шаблон не найден FlowValidationError: Если конфигурация невалидна """ template = get_template(template_name) if template is None: available = ", ".join(sorted(TEMPLATES.keys())) raise ValueError( f"Шаблон '{template_name}' не найден. " f"Доступные шаблоны: {available}" ) customizations = customizations or {} # Валидация customizations for key, value in customizations.items(): if key not in template.customizations: logger.warning( "flows.create.unknown_customization", template=template_name, key=key, ) # Создание через Builder builder = CustomFlowBuilder() builder.with_id(flow_id) builder.with_name(flow_name or template.display_name) builder.with_description(description or template.description) # Добавление агентов из шаблона for agent_config in template.agents: builder.add_agent( agent_type=agent_config["type"], name=agent_config.get("name"), system_prompt=agent_config.get("system_prompt"), input_template=agent_config.get("input_template"), ) # Добавление metadata о шаблоне и кастомизациях builder.with_metadata("template", template_name) builder.with_metadata("category", template.category) if customizations: builder.with_metadata("customizations", customizations) builder.with_metadata("tags", template.tags) # Build and optionally register flow = builder.build(auto_register=auto_register) logger.info( "flows.create.from_template", template=template_name, flow_id=flow_id, agents_count=len(template.agents), ) return flow def create_similar_flow( source_flow_id: str, new_flow_id: str, new_name: str | None = None, modifications: dict[str, Any] | None = None, auto_register: bool = False, ) -> Flow: """ Создать новый flow на основе существующего. Args: source_flow_id: ID исходного flow new_flow_id: ID нового flow new_name: Новое имя modifications: Модификации (add_agents, remove_agents, etc.) auto_register: Автоматически зарегистрировать Returns: Новый Flow instance """ from src.flows.registry import get_flow source_flow = get_flow(source_flow_id) if source_flow is None: raise ValueError(f"Flow '{source_flow_id}' не найден") modifications = modifications or {} # Копируем агентов agents_config = getattr(source_flow, "agents_config", None) if agents_config is None: # Простой flow без кастомных конфигов agents_list = source_flow.agents.copy() # Применяем модификации if "add_agents" in modifications: agents_list.extend(modifications["add_agents"]) if "remove_agents" in modifications: for agent in modifications["remove_agents"]: if agent in agents_list: agents_list.remove(agent) return create_custom_flow( id=new_flow_id, name=new_name or f"{source_flow.name} (modified)", agents=agents_list, description=source_flow.description, auto_register=auto_register, ) # Для custom flows используем Builder builder = CustomFlowBuilder() builder.with_id(new_flow_id) builder.with_name(new_name or f"{source_flow.name} (modified)") builder.with_description(source_flow.description) for agent_config in agents_config: builder.add_agent( agent_type=agent_config.agent_type, name=agent_config.name, system_prompt=agent_config.system_prompt, input_template=agent_config.input_template, ) # Добавляем новые агенты if "add_agents" in modifications: for agent_data in modifications["add_agents"]: if isinstance(agent_data, str): builder.add_agent(agent_type=agent_data) elif isinstance(agent_data, dict): builder.add_agent( agent_type=agent_data["type"], name=agent_data.get("name"), system_prompt=agent_data.get("system_prompt"), ) return builder.build(auto_register=auto_register) # ============================================================================ # YAML Generation # ============================================================================ def generate_yaml_from_template( template_name: str, flow_id: str, flow_name: str | None = None, customizations: dict[str, Any] | None = None, ) -> str: """ Сгенерировать YAML конфиг из шаблона (без создания flow). Полезно для предварительного просмотра и ручного редактирования. Args: template_name: Имя шаблона flow_id: ID flow flow_name: Имя flow customizations: Параметры кастомизации Returns: YAML строка """ import yaml template = get_template(template_name) if template is None: raise ValueError(f"Шаблон '{template_name}' не найден") customizations = customizations or {} data = { "id": flow_id, "name": flow_name or template.display_name, "description": template.description, "agents": template.agents, "metadata": { "template": template_name, "category": template.category, "tags": template.tags, "customizations": customizations, }, } return yaml.dump( data, allow_unicode=True, sort_keys=False, default_flow_style=False, ) # ============================================================================ # CLI Interface # ============================================================================ class Colors: """ANSI escape коды для цветного вывода.""" RESET = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" RED = "\033[31m" GREEN = "\033[32m" YELLOW = "\033[33m" BLUE = "\033[34m" MAGENTA = "\033[35m" CYAN = "\033[36m" BRIGHT_GREEN = "\033[92m" BRIGHT_CYAN = "\033[96m" BRIGHT_MAGENTA = "\033[95m" def _color(text: str, color_code: str) -> str: if not sys.stdout.isatty(): return text return f"{color_code}{text}{Colors.RESET}" def _print_template(template: FlowTemplate) -> None: """Красиво вывести информацию о шаблоне.""" print(_color(f"\n📋 {template.name}", Colors.BOLD + Colors.BRIGHT_MAGENTA)) print(f" {_color('Name:', Colors.DIM)} {template.display_name}") print(f" {_color('Category:', Colors.DIM)} {template.category}") print(f" {_color('Description:', Colors.DIM)} {template.description}") print(f" {_color('Tags:', Colors.DIM)} {', '.join(template.tags)}") print(f" {_color('Agents:', Colors.DIM)}") for agent in template.agents: agent_type = agent["type"] name = agent.get("name", agent_type) print(f" • {name} ({agent_type})") async def cmd_list_templates(args: argparse.Namespace) -> int: """Список доступных шаблонов.""" templates = list_templates(category=args.category) if not templates: print(_color("⚠️ Шаблоны не найдены", Colors.YELLOW)) return 1 print(_color( f"\n{'=' * 70}\nДоступные шаблоны ({len(templates)})\n{'=' * 70}", Colors.BOLD + Colors.BLUE )) # Группировка по категориям by_category: dict[str, list[FlowTemplate]] = {} for template in templates: by_category.setdefault(template.category, []).append(template) for category, category_templates in sorted(by_category.items()): print(_color(f"\n▶ {category.upper()} ({len(category_templates)})", Colors.BOLD + Colors.CYAN)) for template in category_templates: _print_template(template) print() print(_color( "Используй: python -m src.flows.create --template <name> --id <flow_id>", Colors.DIM )) return 0 async def cmd_create(args: argparse.Namespace) -> int: """Создать flow из шаблона.""" try: # Парсим customizations из --set key=value customizations: dict[str, Any] = {} if args.set: for item in args.set: if "=" not in item: print(_color(f"❌ Неверный формат --set: {item}", Colors.RED)) return 1 key, value = item.split("=", 1) # Пробуем преобразовать в JSON try: customizations[key] = json.loads(value) except json.JSONDecodeError: customizations[key] = value # Создаём flow flow = create_from_template( template_name=args.template, flow_id=args.id, flow_name=args.name, description=args.description, customizations=customizations, auto_register=not args.no_register, ) print(_color(f"\n✅ Flow создан успешно!", Colors.BRIGHT_GREEN)) print(f" {_color('ID:', Colors.BOLD)} {flow.id}") print(f" {_color('Name:', Colors.BOLD)} {flow.name}") print(f" {_color('Agents:', Colors.BOLD)} {', '.join(flow.agents)}") if args.no_register: print(_color(" ⚠️ Flow НЕ зарегистрирован (используй --register)", Colors.YELLOW)) else: print(_color(" ✓ Зарегистрирован в registry", Colors.GREEN)) # Сохраняем в YAML если запрошено if args.output: export_flow_to_yaml(flow, args.output) print(_color(f" ✓ Сохранён в {args.output}", Colors.GREEN)) return 0 except ValueError as e: print(_color(f"❌ Ошибка: {e}", Colors.RED)) return 1 except FlowValidationError as e: print(_color(f"❌ Ошибка валидации: {e}", Colors.RED)) return 1 async def cmd_generate_yaml(args: argparse.Namespace) -> int: """Сгенерировать YAML из шаблона.""" try: customizations: dict[str, Any] = {} if args.set: for item in args.set: if "=" not in item: continue key, value = item.split("=", 1) customizations[key] = value yaml_content = generate_yaml_from_template( template_name=args.template, flow_id=args.id, flow_name=args.name, customizations=customizations, ) if args.output: from pathlib import Path Path(args.output).write_text(yaml_content, encoding="utf-8") print(_color(f"✅ YAML сохранён в {args.output}", Colors.BRIGHT_GREEN)) else: print(_color("\n📄 YAML конфигурация:", Colors.BOLD + Colors.BLUE)) print(yaml_content) return 0 except ValueError as e: print(_color(f"❌ Ошибка: {e}", Colors.RED)) return 1 async def cmd_inspect_template(args: argparse.Namespace) -> int: """Детальная информация о шаблоне.""" template = get_template(args.template) if template is None: print(_color(f"❌ Шаблон '{args.template}' не найден", Colors.RED)) return 1 _print_template(template) if template.customizations: print(_color(f"\n {_color('Customizations:', Colors.BOLD)}", Colors.BOLD)) for key, config in template.customizations.items(): default = config.get("default", "None") config_type = config.get("type", "string") desc = config.get("description", "") print(f" • {key}:") print(f" Type: {config_type}") print(f" Default: {default}") if desc: print(f" Description: {desc}") if "options" in config: print(f" Options: {', '.join(config['options'])}") return 0 async def cmd_interactive(args: argparse.Namespace) -> int: """Интерактивное создание flow.""" print(_color("\n🎯 Интерактивный режим создания flow\n", Colors.BOLD + Colors.BLUE)) # Шаг 1: Выбор шаблона templates = list_templates() print(_color("Доступные шаблоны:", Colors.BOLD)) for i, template in enumerate(templates, 1): print(f" {i}. {template.name} — {template.display_name}") while True: try: choice = input("\nВыберите номер шаблона: ").strip() idx = int(choice) - 1 if 0 <= idx < len(templates): selected_template = templates[idx] break print(_color("❌ Неверный номер", Colors.RED)) except (ValueError, KeyboardInterrupt): print(_color("\n⚠️ Отменено", Colors.YELLOW)) return 130 print(_color(f"\n✓ Выбран шаблон: {selected_template.display_name}", Colors.GREEN)) # Шаг 2: Ввод ID try: flow_id = input("\nID flow (snake_case): ").strip() if not flow_id: flow_id = f"{selected_template.name}_{int(asyncio.get_event_loop().time())}" print(_color(f" Используется автоматически: {flow_id}", Colors.DIM)) except KeyboardInterrupt: return 130 # Шаг 3: Ввод имени (опционально) try: flow_name = input(f"Имя flow [{selected_template.display_name}]: ").strip() if not flow_name: flow_name = selected_template.display_name except KeyboardInterrupt: return 130 # Шаг 4: Кастомизации customizations: dict[str, Any] = {} if selected_template.customizations: print(_color("\nПараметры кастомизации:", Colors.BOLD)) for key, config in selected_template.customizations.items(): default = config.get("default", "") desc = config.get("description", key) try: value = input(f" {desc} [{default}]: ").strip() customizations[key] = value if value else default except KeyboardInterrupt: return 130 # Шаг 5: Подтверждение print(_color("\n📋 Итоговая конфигурация:", Colors.BOLD)) print(f" Template: {selected_template.name}") print(f" ID: {flow_id}") print(f" Name: {flow_name}") print(f" Agents: {', '.join(a['type'] for a in selected_template.agents)}") if customizations: print(f" Customizations: {customizations}") try: confirm = input("\nСоздать flow? [Y/n]: ").strip().lower() if confirm and confirm != "y": print(_color("⚠️ Отменено", Colors.YELLOW)) return 0 except KeyboardInterrupt: return 130 # Создание try: flow = create_from_template( template_name=selected_template.name, flow_id=flow_id, flow_name=flow_name, customizations=customizations, auto_register=True, ) print(_color(f"\n✅ Flow '{flow.id}' успешно создан и зарегистрирован!", Colors.BRIGHT_GREEN)) return 0 except Exception as e: print(_color(f"\n❌ Ошибка: {e}", Colors.RED)) return 1 # ============================================================================ # Argument Parser # ============================================================================ def build_parser() -> argparse.ArgumentParser: """Создать парсер аргументов.""" parser = argparse.ArgumentParser( prog="python -m src.flows.create", description="FlowStack Flow Creator — удобное создание flows из шаблонов", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Примеры: # Список всех шаблонов python -m src.flows.create list # Список шаблонов в категории python -m src.flows.create list --category research # Создать flow из шаблона python -m src.flows.create create --template research_deep --id my_research # Создать с кастомизациями python -m src.flows.create create --template research_deep --id my_flow \\ --set language=ru --set focus=medical # Создать и сохранить в YAML python -m src.flows.create create --template research_deep --id my_flow -o flow.yaml # Только сгенерировать YAML (без создания flow) python -m src.flows.create yaml --template research_deep --id my_flow -o flow.yaml # Детали шаблона python -m src.flows.create inspect research_deep # Интерактивный режим python -m src.flows.create interactive """, ) subparsers = parser.add_subparsers(dest="command", help="Команда") # list p_list = subparsers.add_parser("list", help="Список доступных шаблонов") p_list.add_argument("--category", "-c", help="Фильтр по категории") # create p_create = subparsers.add_parser("create", help="Создать flow из шаблона") p_create.add_argument("--template", "-t", required=True, help="Имя шаблона") p_create.add_argument("--id", required=True, help="ID создаваемого flow") p_create.add_argument("--name", "-n", help="Имя flow (опционально)") p_create.add_argument("--description", "-d", help="Описание flow") p_create.add_argument("--set", "-s", action="append", help="Параметр кастомизации (key=value, можно несколько)") p_create.add_argument("--output", "-o", help="Сохранить в YAML файл") p_create.add_argument("--no-register", action="store_true", help="Не регистрировать в registry") # yaml (только генерация YAML) p_yaml = subparsers.add_parser("yaml", help="Сгенерировать YAML без создания flow") p_yaml.add_argument("--template", "-t", required=True, help="Имя шаблона") p_yaml.add_argument("--id", required=True, help="ID flow") p_yaml.add_argument("--name", "-n", help="Имя flow") p_yaml.add_argument("--set", "-s", action="append", help="Кастомизация") p_yaml.add_argument("--output", "-o", help="Сохранить в файл") # inspect p_inspect = subparsers.add_parser("inspect", help="Детали шаблона") p_inspect.add_argument("template", help="Имя шаблона") # interactive subparsers.add_parser("interactive", help="Интерактивное создание") return parser # ============================================================================ # Command Registry # ============================================================================ COMMANDS: dict[str, Callable] = { "list": cmd_list_templates, "create": cmd_create, "yaml": cmd_generate_yaml, "inspect": cmd_inspect_template, "interactive": cmd_interactive, } # ============================================================================ # Main # ============================================================================ async def main() -> int: """Точка входа.""" # Загружаем .env если есть try: from dotenv import load_dotenv load_dotenv() except ImportError: pass structlog.configure( processors=[ structlog.processors.TimeStamper(fmt="iso"), structlog.processors.add_log_level, structlog.dev.ConsoleRenderer(), ], wrapper_class=structlog.make_filtering_bound_logger(20), ) parser = build_parser() args = parser.parse_args() if not args.command: # По умолчанию — интерактивный режим return await cmd_interactive(args) handler = COMMANDS.get(args.command) if handler is None: print(_color(f"❌ Unknown command: {args.command}", Colors.RED)) parser.print_help() return 1 try: return await handler(args) except KeyboardInterrupt: print(_color("\n⚠️ Прервано", Colors.YELLOW)) return 130 except Exception as e: logger.exception("flows.create.main.failed") print(_color(f"❌ Unexpected error: {e}", Colors.RED)) return 1 # ============================================================================ # Public API # ============================================================================ __all__ = [ # Template class "FlowTemplate", # Template management "TEMPLATES", "list_templates", "get_template", "list_categories", "search_templates", # Flow creation "create_from_template", "create_similar_flow", # YAML generation "generate_yaml_from_template", ] if __name__ == "__main__": sys.exit(asyncio.run(main()))