347 lines
17 KiB
Python
347 lines
17 KiB
Python
import functools
|
||
import importlib.util
|
||
import inspect
|
||
import subprocess
|
||
import json
|
||
import os
|
||
from langchain_core.tools import StructuredTool
|
||
from langgraph.prebuilt import create_react_agent
|
||
from langchain_openai import ChatOpenAI
|
||
from llm_client import MODELS
|
||
|
||
import importlib.util
|
||
import inspect
|
||
import subprocess
|
||
import json
|
||
import os
|
||
import sys
|
||
from typing import Any, get_type_hints
|
||
from pydantic import BaseModel, Field, create_model
|
||
|
||
from mcp_tools import fetch_mcp_tools
|
||
|
||
def build_react_agent(model_name: str, temperature: float, max_tokens: int, agency_mode: bool, obsidian_settings: dict, debug_callback=None):
|
||
"""Инициализация ядра на базе LangGraph."""
|
||
m_cfg = MODELS.get(model_name)
|
||
llm = ChatOpenAI(
|
||
api_key=m_cfg["apiKey"],
|
||
base_url=m_cfg["apiBase"],
|
||
model_name=m_cfg["model_name"],
|
||
temperature=temperature,
|
||
max_tokens=max_tokens if max_tokens else None,
|
||
streaming=True
|
||
)
|
||
tools = get_dynamic_tools(obsidian_settings, debug_callback) if agency_mode else []
|
||
return create_react_agent(llm, tools=tools)
|
||
|
||
# Специфическая Pydantic-модель для инструментов без скрипта
|
||
class PurePromptSchema(BaseModel):
|
||
context: str = Field(description="Контекст, данные или текст, к которому нужно применить эту инструкцию/промпт.")
|
||
|
||
class SystemHelpSchema(BaseModel):
|
||
dummy: str = Field(default="", description="Необязательный параметр, оставь пустым.")
|
||
|
||
def create_subprocess_tool(script_path: str, method_name: str,
|
||
hidden_prompt: str, debug_callback=None):
|
||
"""Вид 2: Инструмент со скриптом + пост-процессинг."""
|
||
|
||
script_dir = os.path.dirname(os.path.abspath(script_path))
|
||
added_to_path = False
|
||
if script_dir not in sys.path:
|
||
sys.path.insert(0, script_dir)
|
||
added_to_path = True
|
||
|
||
try:
|
||
spec = importlib.util.spec_from_file_location("user_dynamic_tool", script_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
if not hasattr(module, method_name):
|
||
error_msg = f"в файле '{script_path}' не найден метод '{method_name}'"
|
||
print(f"❌ {error_msg}")
|
||
raise AttributeError(error_msg)
|
||
|
||
func = getattr(module, method_name)
|
||
docstring = inspect.getdoc(func) or f"Выполняет скрипт {method_name}"
|
||
|
||
# ← КЛЮЧЕВОЕ: строим схему ДО того, как потеряем оригинальную сигнатуру
|
||
args_schema = _build_pydantic_schema_from_func(func)
|
||
|
||
except Exception as e:
|
||
raise RuntimeError(f"Ошибка чтения сигнатуры {script_path}: {e}")
|
||
finally:
|
||
# Убираем путь сразу после импорта и получения функции
|
||
if added_to_path:
|
||
sys.path.remove(script_dir)
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(**kwargs: Any) -> str:
|
||
input_json = json.dumps(kwargs, indent=2, ensure_ascii=False)
|
||
runner_code = f"""
|
||
import json
|
||
import sys
|
||
import importlib.util
|
||
import os
|
||
|
||
# ПРИНУДИТЕЛЬНО устанавливаем кодировку UTF-8 для потоков вывода и ошибок, чтобы могло выводить эмодзи, например, без ошибок
|
||
if hasattr(sys.stdout, 'reconfigure'):
|
||
sys.stdout.reconfigure(encoding='utf-8')
|
||
if hasattr(sys.stderr, 'reconfigure'):
|
||
sys.stderr.reconfigure(encoding='utf-8')
|
||
|
||
# Добавляем папку со скриптом в sys.path, чтобы работали локальные импорты (например, norminv)
|
||
path_to_add = os.path.dirname(os.path.abspath(r'{script_path}'))
|
||
original_path = sys.path.copy()
|
||
sys.path.insert(0, path_to_add)
|
||
|
||
try:
|
||
spec = importlib.util.spec_from_file_location('sub_module', r'{script_path}')
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
kwargs = json.loads(sys.stdin.read())
|
||
result = getattr(module, '{method_name}')(**kwargs)
|
||
print(result)
|
||
except Exception as e:
|
||
print(f"SCRIPT RUNTIME ERROR: {{e}}", file=sys.stderr)
|
||
sys.exit(1)
|
||
finally:
|
||
# Возвращаем sys.path в исходное состояние (чистота внутри процесса)
|
||
sys.path = original_path
|
||
"""
|
||
try:
|
||
process = subprocess.run(
|
||
[sys.executable, "-c", runner_code],
|
||
input=json.dumps(kwargs),
|
||
capture_output=True,
|
||
text=True,
|
||
encoding='utf-8',
|
||
check=True
|
||
)
|
||
raw_output = process.stdout.strip()
|
||
|
||
# ВЫЗЫВАЕМ КОЛБЭК, если он передан
|
||
if debug_callback:
|
||
debug_callback(input_json, raw_output)
|
||
|
||
raw_stripped = raw_output.strip() if raw_output else ""
|
||
prompt_stripped = hidden_prompt.strip() if hidden_prompt else ""
|
||
|
||
if raw_stripped and prompt_stripped:
|
||
# Случай 1: Есть и результат скрипта, и инструкция — полный вариант
|
||
result = (
|
||
f"Инструмент вернул следующий результат:\n"
|
||
f"<tool_output>\n{raw_output}\n</tool_output>\n\n"
|
||
f"---\n"
|
||
f"Инструкция по оформлению финального ответа пользователю:\n"
|
||
f"<instruction>\n{hidden_prompt}\n</instruction>"
|
||
)
|
||
|
||
elif raw_stripped and not prompt_stripped:
|
||
# Случай 2: Есть результат, нет инструкции
|
||
result = (
|
||
f"Инструмент успешно выполнен. Передай пользователю следующий результат "
|
||
f"ДОСЛОВНО, без изменений, сокращений и дополнений:\n\n"
|
||
f"<tool_output>\n{raw_output}\n</tool_output>"
|
||
)
|
||
|
||
elif not raw_stripped and prompt_stripped:
|
||
# Случай 3: Скрипт вернул пустоту (заглушка/нет данных) — только промпт как инструкция
|
||
result = prompt_stripped
|
||
""" result = (
|
||
f"Скрипт не вернул данных. Выполни следующую инструкцию:\n\n{hidden_prompt}"
|
||
)"""
|
||
|
||
else:
|
||
# Случай 4: Оба пустые — сообщаем агенту честно
|
||
result = (
|
||
"Инструмент не вернул никаких данных и не содержит инструкций. "
|
||
"Сообщи пользователю, что инструмент настроен некорректно."
|
||
)
|
||
|
||
return result
|
||
|
||
except subprocess.CalledProcessError as e:
|
||
# Даже если скрипт упал, мы отдаем ошибку + промпт, чтобы агент мог обработать её по инструкции
|
||
error_msg = f"ОШИБКА ИСПОЛНЕНИЯ (Код {e.returncode}):\nSTDOUT: {e.stdout}\nSTDERR: {e.stderr}"
|
||
if hidden_prompt:
|
||
return f"{error_msg}\n\nИНСТРУКЦИЯ НА СЛУЧАЙ ОШИБКИ:\n{hidden_prompt}"
|
||
return error_msg
|
||
|
||
# Явно передаём args_schema — LLM будет видеть реальную типизацию
|
||
return StructuredTool.from_function(
|
||
func=wrapper,
|
||
name=func.__name__,
|
||
description=docstring,
|
||
args_schema=args_schema, # ← вместо infer_schema=True
|
||
)
|
||
|
||
def get_dynamic_tools(obsidian_settings: dict, debug_callback=None):
|
||
"""Собирает инструменты двух видов на основе настроек Obsidian."""
|
||
tools = []
|
||
|
||
# Инициализация внутренних инструментов (скриптов)
|
||
custom_tools = obsidian_settings.get("customTools", [])
|
||
|
||
for ct in custom_tools:
|
||
name = ct.get("name")
|
||
desc = ct.get("description")
|
||
# Поддерживаем оба ключа, если в JS/TS они не раскрылись
|
||
hidden_prompt = ct.get("hiddenPromptExpanded") or ct.get("hiddenPrompt", "")
|
||
|
||
if not name:
|
||
continue # Инструмент без имени создать нельзя
|
||
|
||
# Достаем путь и метод из отдельных полей, присланных фронтендом
|
||
script_path = ct.get("scriptPath", "")
|
||
method_name = ct.get("methodName") or "main"
|
||
|
||
# ПРОВЕРКА: Есть ли физический скрипт?
|
||
is_valid_script = script_path and os.path.exists(script_path)
|
||
|
||
if not is_valid_script:
|
||
# ----------------------------------------------------
|
||
# ВИД 1: Чистый промпт (Инструмент-инструкция / Системная подсказка)
|
||
# ----------------------------------------------------
|
||
def create_pure_prompt_wrapper(prompt_text: str):
|
||
# Функция принимает context, который ИИ обязан туда положить
|
||
def pure_prompt_wrapper(context: str) -> str:
|
||
return (
|
||
f"ВХОДНЫЕ ДАННЫЕ ОТ ПОЛЬЗОВАТЕЛЯ/МОДЕЛИ:\n{context}\n\n"
|
||
f"ИНСТРУКЦИЯ ПО ОБРАБОТКЕ И ПРИЛИЗЫВАНИЮ (ОБЯЗАТЕЛЬНО К ИСПОЛНЕНИЮ):\n{prompt_text}"
|
||
)
|
||
return pure_prompt_wrapper
|
||
|
||
tool = StructuredTool.from_function(
|
||
func=create_pure_prompt_wrapper(hidden_prompt),
|
||
name=name,
|
||
description=desc or "Инструмент для применения специфических правил форматирования или инструкций.",
|
||
args_schema=PurePromptSchema # Жестко навязываем схему с аргументом context!
|
||
)
|
||
tools.append(tool)
|
||
print(f"🛠 Инструмент-промпт '{tool.name}' (без скрипта) успешно инициализирован.")
|
||
|
||
else:
|
||
# ----------------------------------------------------
|
||
# ВИД 2: Скрипт + Прилизывание
|
||
# ----------------------------------------------------
|
||
try:
|
||
tool = create_subprocess_tool(script_path, method_name, hidden_prompt, debug_callback)
|
||
tool.name = name # Приоритет имени из UI Obsidian
|
||
if desc:
|
||
tool.description = desc
|
||
tools.append(tool)
|
||
print(f"🛠 Инструмент-скрипт '{tool.name}' успешно инициализирован.")
|
||
except Exception as e:
|
||
print(f"❌ Ошибка генерации инструмента-скрипта '{name}': {e}")
|
||
|
||
def run_system_help(dummy: str = "") -> str:
|
||
return _generate_commands_list(custom_tools)
|
||
|
||
help_tool = StructuredTool.from_function(
|
||
func=run_system_help,
|
||
name="show_available_commands",
|
||
description="""Используй этот инструмент, если пользователь спрашивает "Что ты умеешь?",
|
||
"Покажи help", "Какие есть команды" или просит список возможностей.""",
|
||
args_schema=SystemHelpSchema
|
||
)
|
||
tools.append(help_tool)
|
||
|
||
vault_cwd = obsidian_settings.get("vaultAbsolutePath")
|
||
|
||
# Инициализация внутренних MCP серверов
|
||
mcp_servers = obsidian_settings.get("mcpServers", [])
|
||
for mcp_server in mcp_servers:
|
||
# Пропускаем отключенные серверы
|
||
if not mcp_server.get("isEnabled", True):
|
||
continue
|
||
|
||
mcp_tools = fetch_mcp_tools(mcp_server, debug_callback, vault_cwd=vault_cwd)
|
||
tools.extend(mcp_tools)
|
||
|
||
return tools
|
||
|
||
def _build_pydantic_schema_from_func(func) -> type[BaseModel]:
|
||
"""
|
||
Динамически строит Pydantic BaseModel из сигнатуры функции.
|
||
Используется для корректной передачи схемы в StructuredTool,
|
||
чтобы LLM видела реальные типы аргументов, а не **kwargs.
|
||
"""
|
||
sig = inspect.signature(func)
|
||
|
||
# Пробуем получить аннотации типов (включая from __future__ import annotations)
|
||
try:
|
||
hints = get_type_hints(func)
|
||
except Exception:
|
||
hints = {}
|
||
|
||
fields = {}
|
||
for param_name, param in sig.parameters.items():
|
||
if param_name in ("self", "cls"):
|
||
continue
|
||
if param.kind in (
|
||
inspect.Parameter.VAR_POSITIONAL, # *args
|
||
inspect.Parameter.VAR_KEYWORD, # **kwargs
|
||
):
|
||
continue
|
||
|
||
# Тип параметра
|
||
annotation = hints.get(param_name, Any)
|
||
|
||
# Дефолтное значение
|
||
if param.default is inspect.Parameter.empty:
|
||
# Обязательный параметр
|
||
fields[param_name] = (annotation, Field(...))
|
||
else:
|
||
fields[param_name] = (annotation, Field(default=param.default))
|
||
|
||
if not fields:
|
||
# Фоллбек: если функция не имеет параметров или все **kwargs —
|
||
# создаём схему с одним полем, чтобы LLM хоть что-то передала
|
||
fields["kwargs"] = (dict, Field(default_factory=dict,
|
||
description="Аргументы функции"))
|
||
|
||
model = create_model(
|
||
f"{func.__name__}_Schema",
|
||
**fields
|
||
)
|
||
return model
|
||
|
||
def _generate_commands_list(custom_tools: list[dict[str, Any]]) -> str:
|
||
"""
|
||
Генерирует форматированный список доступных команд и их описаний
|
||
для использования в инструменте help.
|
||
"""
|
||
if not custom_tools:
|
||
return "В данный момент нет доступных дополнительных команд. Ты работаешь как обычный собеседник."
|
||
|
||
tool_raw_list = ["Я могу использовать следующие инструменты/команды:\n"]
|
||
for ct in custom_tools:
|
||
name = ct.get("name")
|
||
desc = ct.get("description", "Описание отсутствует")
|
||
if name:
|
||
tool_raw_list.append(f"- **{name}**: {desc}")
|
||
|
||
# Приаттачиваем инструкцию формата вывода (hidden prompt для системы)
|
||
prompt = f"""
|
||
СПИСОК ДОСТУПНЫХ ИНСТРУМЕНТОВ:
|
||
{tool_raw_list}
|
||
|
||
ТВОЯ ЗАДАЧА:
|
||
Сгруппируй инструменты по уровням абстракции и выведи их в формате Markdown-таблиц.
|
||
|
||
ИСПОЛЬЗУЙ СЛЕДУЮЩИЕ КАТЕГОРИИ:
|
||
1. 🧩 [Микро-уровень]: Атомарные сущности, единичные объекты, конкретные детали или расчеты.
|
||
2. 🏗️ [Мезо-уровень]: Процессы, сложные узлы, локальные системы, взаимодействия или окружение.
|
||
3. 🗺️ [Макро-уровень]: Глобальные структуры, архитектура системы, высокоуровневая логика или долгосрочные стратегии.
|
||
4. ⚙️ [Мета-уровень]: Управление сессией, диагностика, системная помощь, логирование или работа с контекстом.
|
||
|
||
ФОРМАТ ВЫВОДА:
|
||
Для каждой категории создай заголовок и таблицу:
|
||
| Команда | Функциональное назначение |
|
||
|:---|:---|
|
||
| **название_команды** | Суть работы (1 краткое предложение) |
|
||
|
||
Не добавляй никаких приветствий или пояснений вне блока кода. Будь предельно лаконичен.
|
||
"""
|
||
return prompt |