Human-in-the-Loop через middleware: solution.py

solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py

@@ -0,0 +1,126 @@
+import sys
+from typing import List, Dict, Any
+
+# ────────────────────────  Импорт библиотек  ────────────────────────
+try:
+    from langchain.agents import create_agent
+    from langchain.agents.middleware import HumanInTheLoopMiddleware
+    from langgraph.checkpoint.memory import MemorySaver
+    from langgraph.types import Command
+except Exception as exc:  # pragma: no cover
+    print("Не удалось импортировать необходимые библиотеки. Убедитесь, что установлены langchain и langgraph.", file=sys.stderr)
+    raise exc
+
+# ────────────────────────  Настройка модели и инструментов  ────────────────────────
+# Для примера используем простую модель из LangChain (OpenAI). Если у вас нет ключа,
+# замените на DummyLLM или другую доступную модель.
+try:
+    from langchain_community.llms import OpenAI
+
+    llm = OpenAI(temperature=0.7)
+except Exception:  # pragma: no cover
+    # В случае отсутствия OpenAI создаём заглушку, которая просто возвращает фиксированный ответ
+    class DummyLLM:
+        def __call__(self, *args, **kwargs):
+            return "Dummy response"
+
+    llm = DummyLLM()
+
+# Пример простого инструмента: получение погоды (заглушка)
+def get_weather(location: str) -> str:
+    """Возвращает фиктивную погоду для указанного места."""
+    return f"Погода в {location} сегодня солнечная."
+
+tools = [get_weather]
+
+# ────────────────────────  Создание агента с HumanInTheLoopMiddleware  ────────────────────────
+memory = MemorySaver()
+
+agent = create_agent(
+    model=llm,
+    tools=tools,
+    system_prompt="Ты полезный ассистент",
+    middleware=[
+        HumanInTheLoopMiddleware(
+            interrupt_on={
+                "get_weather": True,  # все решения: approve, edit, reject
+            },
+            description_prefix="Подтвердите вызов инструмента",
+        ),
+    ],
+    checkpointer=memory,
+)
+
+# ────────────────────────  Функция взаимодействия с пользователем  ────────────────────────
+def ask_and_run(user_input: Dict[str, Any], config: Dict[str, Any]) -> None:
+    """
+    Выполняет запрос к агенту, обрабатывает паузу Human‑in‑the‑Loop и возвращает результат.
+    """
+    # Первый вызов агента
+    result = agent.invoke(user_input, config=config)
+
+    while "__interrupt__" in result:
+        interrupt_value = result["__interrupt__"][0].value
+        action_requests: List[Dict[str, Any]] = interrupt_value.get("action_requests", [])
+        review_configs: List[Dict[str, Any]] = interrupt_value.get("review_configs", [])
+
+        decisions: List[Dict[str, Any]] = []
+
+        # Для каждого запроса инструмента запрашиваем у пользователя решение
+        for idx, action in enumerate(action_requests):
+            name = action.get("name")
+            args = action.get("args", {})
+            description = action.get("description")
+
+            print("\n--- Запрос на подтверждение ---")
+            print(f"Инструмент: {name}")
+            print(f"Аргументы: {args}")
+            if description:
+                print(f"Описание: {description}")
+
+            # Определяем допустимые решения
+            allowed_decisions = ["approve", "reject"]
+            for cfg in review_configs:
+                if cfg.get("name") == name:
+                    allowed_decisions = cfg.get("allowed_decisions", allowed_decisions)
+                    break
+
+            # Запрашиваем ввод от пользователя
+            while True:
+                choice = input(f"Выберите действие ({'/'.join(allowed_decisions)}): ").strip().lower()
+                if choice in ("a", "approve") and "approve" in allowed_decisions:
+                    decisions.append({"type": "approve"})
+                    break
+                elif choice in ("r", "reject") and "reject" in allowed_decisions:
+                    msg = input("Введите причину отказа: ").strip() or "Нет причины"
+                    decisions.append({"type": "reject", "message": msg})
+                    break
+                else:
+                    print(f"Недопустимый выбор. Допустимые варианты: {allowed_decisions}")
+
+        # Возобновляем работу агента с полученными решениями
+        result = agent.invoke(Command(resume={"decisions": decisions}), config=config)
+
+    # После завершения (без паузы) выводим финальный ответ
+    if "messages" in result:
+        for msg in result["messages"]:
+            role = msg.get("role", "")
+            content = msg.get("content", "")
+            print(f"\n{role.capitalize()}: {content}")
+
+# ────────────────────────  Основной цикл чата  ────────────────────────
+def main():
+    config = {"configurable": {"thread_id": "session-1"}}
+    print("Привет! Я ассистент. Введите 'exit' для завершения.")
+    while True:
+        user_text = input("\nВы: ").strip()
+        if user_text.lower() == "exit":
+            print("До свидания!")
+            break
+        ask_and_run(
+            {"messages": [{"role": "human", "content": user_text}]},
+            config=config,
+        )
+
+if __name__ == "__main__":
+    main()