add main.py

main.py

@@ -0,0 +1,120 @@
+"""Human-in-the-Loop через HumanInTheLoopMiddleware (LangChain)."""
+from __future__ import annotations
+
+import os
+
+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()
+
+# LLM: OpenRouter из .env или локальный LM Studio
+llm = ChatOpenAI(
+    model=os.getenv("OPENAI_MODEL", "openai/gpt-oss-20b:free"),
+    base_url=os.getenv("OPENAI_BASE_URL", "https://openrouter.ai/api/v1"),
+    api_key=os.getenv("OPENAI_API_KEY", "fake"),
+    temperature=0.0,
+)
+
+
+@tool
+def get_weather(city: str, date: str = "сегодня") -> str:
+    """Получить погоду в городе на указанную дату."""
+    return f"В {city} на {date}: около +5°C, облачно, без осадков."
+
+
+memory = MemorySaver()
+
+agent = create_agent(
+    model=llm,
+    tools=[get_weather],
+    system_prompt="Ты полезный ассистент",
+    middleware=[
+        HumanInTheLoopMiddleware(
+            interrupt_on={
+                "get_weather": {"allowed_decisions": ["approve", "reject"]},
+            },
+            description_prefix="Подтвердите вызов инструмента",
+        ),
+    ],
+    checkpointer=memory,
+)
+
+
+def _print_action_requests(action_requests: list[dict]) -> None:
+    print("\n--- Подтверждение ---")
+    for idx, action in enumerate(action_requests, start=1):
+        print(f"[{idx}] Инструмент: {action.get('name')}")
+        print(f"    Аргументы: {action.get('args')}")
+        description = action.get("description")
+        if description:
+            print(f"    Описание: {description}")
+
+
+def _ask_decision() -> dict:
+    while True:
+        choice = input("a = approve, r = reject: ").strip().lower()
+        if choice in ("a", "approve"):
+            return {"type": "approve"}
+        if choice in ("r", "reject"):
+            message = input("Сообщение для агента (причина отказа): ").strip()
+            return {"type": "reject", "message": message or "Отклонено пользователем"}
+        print("Введите 'a' или 'r'.")
+
+
+def _collect_decisions(action_requests: list[dict]) -> list[dict]:
+    _print_action_requests(action_requests)
+    return [_ask_decision() for _ in action_requests]
+
+
+def _extract_interrupt(result: dict) -> tuple[list[dict], list[dict]] | None:
+    if "__interrupt__" not in result:
+        return None
+    interrupt_value = result["__interrupt__"][0].value
+    return interrupt_value.get("action_requests", []), interrupt_value.get("review_configs", [])
+
+
+def run_with_hitl(user_message: str, thread_id: str = "hitl-middleware-session-1") -> str:
+    """Запуск агента с циклом подтверждения инструментов."""
+    config = {"configurable": {"thread_id": thread_id}}
+    result = agent.invoke(
+        {"messages": [{"role": "human", "content": user_message}]},
+        config=config,
+    )
+
+    while True:
+        interrupted = _extract_interrupt(result)
+        if not interrupted:
+            break
+        action_requests, _review_configs = interrupted
+        decisions = _collect_decisions(action_requests)
+        result = agent.invoke(Command(resume={"decisions": decisions}), config=config)
+
+    messages = result.get("messages", [])
+    if not messages:
+        return ""
+    return str(messages[-1].content)
+
+
+def main() -> None:
+    print("Human-in-the-Loop (middleware). Введите 'exit' для выхода.\n")
+    config_thread = "hitl-middleware-session-1"
+
+    while True:
+        user_input = input("Вы: ").strip()
+        if user_input.lower() in ("exit", "quit", "q"):
+            break
+        if not user_input:
+            continue
+
+        answer = run_with_hitl(user_input, thread_id=config_thread)
+        print(f"\nАгент: {answer}\n")
+
+
+if __name__ == "__main__":
+    main()