task-69a86305c46fd26feae6bcaa/main.py

"""Human-in-the-Loop через HumanInTheLoopMiddleware (LangChain)."""
from __future__ import annotations

import os
from typing import Any

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

load_dotenv()

BROJS_INFERENCE_URL = "https://platform.brojs.ru/jrnl-bh/api/inference/v1"
DEFAULT_MODEL = "openai/gpt-oss-20b:free"


def _api_key() -> str:
    return (
        os.getenv("OPENAI_API_KEY")
        or os.getenv("JOURNAL_MCP_PAT")
        or os.getenv("JOURNAL_TOKEN")
        or ""
    )


def _base_url() -> str:
    if os.getenv("OPENAI_BASE_URL"):
        return os.environ["OPENAI_BASE_URL"]
    if os.getenv("OPENAI_API_KEY"):
        return os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
    return BROJS_INFERENCE_URL


def _model() -> str:
    return os.getenv("OPENAI_MODEL") or os.getenv("OPENROUTER_MODEL") or DEFAULT_MODEL


def build_llm() -> ChatOpenAI:
    return ChatOpenAI(
        model=_model(),
        base_url=_base_url(),
        api_key=_api_key(),
        temperature=0.0,
    )


@tool
def get_weather(city: str, date: str = "сегодня") -> str:
    """Получить погоду в городе на указанную дату."""
    return f"В {city} на {date}: около +5°C, облачно, без осадков."


def build_agent():
    """Агент с HITL-middleware — точка входа для автопроверки."""
    memory = MemorySaver()
    return create_agent(
        model=build_llm(),
        tools=[get_weather],
        system_prompt=(
            "Ты полезный ассистент. Для вопросов о погоде вызывай get_weather."
        ),
        middleware=[
            HumanInTheLoopMiddleware(
                interrupt_on={
                    "get_weather": {"allowed_decisions": ["approve", "reject"]},
                },
                description_prefix="Подтвердите вызов инструмента",
            ),
        ],
        checkpointer=memory,
    )


def _has_interrupt(result: Any) -> bool:
    if isinstance(result, dict):
        return bool(result.get("__interrupt__"))
    return bool(getattr(result, "interrupts", None))


def _interrupt_payload(result: Any) -> dict[str, Any]:
    if isinstance(result, dict) and result.get("__interrupt__"):
        item = result["__interrupt__"][0]
        value = item.value if hasattr(item, "value") else item
        return dict(value)
    interrupts = getattr(result, "interrupts", None)
    if interrupts:
        item = interrupts[0]
        value = item.value if hasattr(item, "value") else item
        return dict(value)
    return {}


def _action_args(action: dict[str, Any]) -> dict[str, Any]:
    return action.get("args") or action.get("arguments") or {}


def _prompt_decisions(
    action_requests: list[dict[str, Any]],
    review_configs: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    allowed_by_tool = {
        cfg.get("action_name", ""): cfg.get("allowed_decisions", ["approve", "reject"])
        for cfg in review_configs
    }
    decisions: list[dict[str, Any]] = []

    for action in action_requests:
        name = action.get("name", "")
        args = _action_args(action)
        allowed = allowed_by_tool.get(name, ["approve", "reject"])

        print("\n--- Подтверждение ---")
        print(f"Инструмент: {name}")
        print(f"Аргументы: {args}")
        if action.get("description"):
            print(f"Описание: {action['description']}")

        while True:
            choice = input("a = approve, r = reject: ").strip().lower()
            if choice in ("a", "approve") and "approve" in allowed:
                decisions.append({"type": "approve"})
                break
            if choice in ("r", "reject") and "reject" in allowed:
                message = input("Сообщение для агента (причина отказа): ").strip()
                decision: dict[str, Any] = {"type": "reject"}
                if message:
                    decision["message"] = message
                decisions.append(decision)
                break
            print("Неверный ввод. Допустимо: a (approve) или r (reject).")

    return decisions


def run_with_hitl(
    agent,
    user_text: str,
    config: dict[str, Any],
    *,
    auto_approve: bool = False,
) -> str:
    """Запуск агента с циклом подтверждения инструментов."""
    result = agent.invoke(
        {"messages": [{"role": "human", "content": user_text}]},
        config=config,
    )

    while _has_interrupt(result):
        payload = _interrupt_payload(result)
        action_requests = payload.get("action_requests", [])
        review_configs = payload.get("review_configs", [])

        if auto_approve:
            decisions = [{"type": "approve"} for _ in action_requests]
        else:
            decisions = _prompt_decisions(action_requests, review_configs)

        result = agent.invoke(
            Command(resume={"decisions": decisions}),
            config=config,
        )

    messages = result.get("messages", []) if isinstance(result, dict) else []
    if not messages:
        return "(агент завершил без текстового ответа)"

    last = messages[-1]
    content = getattr(last, "content", None)
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = [p.get("text", "") for p in content if isinstance(p, dict)]
        return "".join(parts) or str(content)
    return str(content)


def main() -> None:
    agent = build_agent()
    config = {"configurable": {"thread_id": "hitl-middleware-session-1"}}

    print("Human-in-the-Loop (middleware). Введите 'exit' для выхода.\n")

    while True:
        user_text = input("Вы: ").strip()
        if not user_text:
            continue
        if user_text.lower() in {"exit", "quit", "q", "выход"}:
            break

        answer = run_with_hitl(agent, user_text, config)
        print(f"\nАгент: {answer}\n")


if __name__ == "__main__":
    main()