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

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

@@ -1,46 +1,67 @@
-"""
+#!/usr/bin/env python3
-Human-in-the-Loop через middleware
+# -*- coding: utf-8 -*-
 
-Стек строго по условию задания:
-  - langchain.agents.create_agent
-  - langchain.agents.middleware.HumanInTheLoopMiddleware
-  - langgraph.checkpoint.memory.MemorySaver
-  - langgraph.types.Command
-  - result["__interrupt__"] для проверки паузы
-  - Command(resume={"decisions": [...]}) для возобновления
 """
+solution.py
+
+Пример использования HumanInTheLoopMiddleware в LangChain Agent.
+При каждом вызове инструмента агент останавливается и запрашивает у пользователя подтверждение
+(approve / reject). После принятия решения выполнение возобновляется через Command.
+
+Требования:
+    pip install langchain langgraph openai rich
+"""
+
+import json
+from typing import Any, Dict, List
+
+# --------------------------------------------------------------------------- #
+# 1. Подключаем необходимые модули LangChain и LangGraph
+# --------------------------------------------------------------------------- #
 
 from langchain.agents import create_agent
 from langchain.agents.middleware import HumanInTheLoopMiddleware
-from langchain_openai import ChatOpenAI
-from langchain.tools import tool
 from langgraph.checkpoint.memory import MemorySaver
 from langgraph.types import Command
+from langchain.tools import tool
 
-# ── Модель ───────────────────────────────────────────────────────────────────
+# --------------------------------------------------------------------------- #
-# ChatOpenAI поддерживает вызов инструментов (tool calling), что необходимо агенту.
+# 2. Определяем простой инструмент get_weather
-# Ключ читается из переменной окружения OPENAI_API_KEY.
+# --------------------------------------------------------------------------- #
-llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
-# ── Инструмент ───────────────────────────────────────────────────────────────
+@tool("get_weather", "Получить погоду в указанном городе и дате")
-@tool
+def get_weather(city: str, date: str) -> str:
-def get_weather(city: str, date: str = "сегодня") -> str:
+    """
-    """Возвращает погоду для указанного города на заданную дату."""
+    Возвращает фиктивную информацию о погоде.
-    return f"В городе {city} {date}: солнечно, +22 °C."
+    В реальном проекте здесь можно подключиться к API погоды.
+    """
+    return f"Погода в {city} на {date}: солнечно, 25°C."
+
+# --------------------------------------------------------------------------- #
+# 3. Создаём LLM (используем OpenAI GPT-4o-mini как пример)
+# --------------------------------------------------------------------------- #
+
+from langchain_openai import ChatOpenAI
+
+llm = ChatOpenAI(
+    model="gpt-4o-mini",
+    temperature=0,
+)
+
+# --------------------------------------------------------------------------- #
+# 4. Настраиваем агент с HumanInTheLoopMiddleware
+# --------------------------------------------------------------------------- #
 
-# ── Агент с HumanInTheLoopMiddleware ─────────────────────────────────────────
-# checkpointer обязателен: без него пауза не сохраняется
 memory = MemorySaver()
 
 agent = create_agent(
     model=llm,
     tools=[get_weather],
-    system_prompt="Ты полезный ассистент.",
+    system_prompt="Ты полезный ассистент, помогающий пользователю.",
     middleware=[
         HumanInTheLoopMiddleware(
             interrupt_on={
-                "get_weather": True,  # все решения: approve, edit, reject
+                "get_weather": True,  # разрешаем все решения
-                # "get_weather": {"allowed_decisions": ["approve", "reject"]},  # без edit
             },
             description_prefix="Подтвердите вызов инструмента",
         ),
@@ -48,122 +69,89 @@ agent = create_agent(
     checkpointer=memory,
 )
 
-# ── Основной чат-цикл ────────────────────────────────────────────────────────
+# --------------------------------------------------------------------------- #
-def main():
+# 5. Функция для получения решений от пользователя
-    # thread_id обязателен — привязывает состояние к одной сессии,
+# --------------------------------------------------------------------------- #
-    # благодаря чему пауза сохраняется и выполнение можно возобновить
-    config = {"configurable": {"thread_id": "сессия-1"}}
 
-    print("Привет! Я ассистент. Введите 'exit' для выхода.")
+def ask_decisions(action_requests: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """
+    Для каждого запроса к инструменту выводим информацию и запрашиваем у пользователя решение.
+    Возвращаем список словарей с решениями в том же порядке, что и action_requests.
+    """
+    decisions = []
+    print("\n--- Подтверждение вызова инструмента ---")
+    for idx, req in enumerate(action_requests):
+        name = req.get("name", "неизвестный инструмент")
+        args = req.get("args", {})
+        description = req.get("description", "")
+
+        print(f"\n{idx + 1}. Инструмент: {name}")
+        if description:
+            print(f"   Описание: {description}")
+        print(f"   Аргументы: {json.dumps(args, ensure_ascii=False)}")
+
+        while True:
+            choice = input("a = approve, r = reject: ").strip().lower()
+            if choice == "a":
+                decisions.append({"type": "approve"})
+                break
+            elif choice == "r":
+                msg = input(
+                    "Введите причину отказа (можно оставить пустой): "
+                ).strip()
+                decisions.append({"type": "reject", "message": msg})
+                break
+            else:
+                print("Неверный ввод. Пожалуйста, введите 'a' или 'r'.")
+    return decisions
+
+# --------------------------------------------------------------------------- #
+# 6. Основной цикл взаимодействия с агентом
+# --------------------------------------------------------------------------- #
+
+def main() -> None:
+    """
+    Запускает чат-цикл: пользователь вводит сообщение,
+    агент обрабатывает его, при необходимости запрашивает подтверждение.
+    После завершения выводится финальный ответ агента.
+    """
+    thread_id = "session-1"
+    config = {"configurable": {"thread_id": thread_id}}
+
+    print("Привет! Я ассистент. Введите ваш запрос (или 'выход' для завершения).")
 
     while True:
-        try:
+        user_msg = input("\nВы: ").strip()
-            user_text = input("\nВы: ").strip()
+        if user_msg.lower() in ("выход", "quit", "exit"):
-        except (EOFError, KeyboardInterrupt):
-            print("\nДо свидания!")
-            break
-
-        if not user_text:
-            continue
-        if user_text.lower() == "exit":
             print("До свидания!")
             break
 
-        # ── Первый вызов агента ───────────────────────────────────────────
+        # Первый вызов агента
         result = agent.invoke(
-            {"messages": [{"role": "human", "content": user_text}]},
+            {"messages": [{"role": "human", "content": user_msg}]},
             config=config,
         )
 
-        # ── Цикл обработки прерываний ─────────────────────────────────────
+        # Цикл подтверждений
-        # Пока агент приостановлен для подтверждения — показываем действие,
-        # собираем решение пользователя и возобновляем через Command.
         while "__interrupt__" in result:
             interrupt_value = result["__interrupt__"][0].value
             action_requests = interrupt_value.get("action_requests", [])
-            review_configs  = interrupt_value.get("review_configs", [])
+            # review_configs не используется в этом примере, но можно вывести при желании
 
-            decisions = []
+            decisions = ask_decisions(action_requests)
 
-            for action in 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 = ["approve", "reject"]
-                for cfg in review_configs:
-                    if cfg.get("name") == name:
-                        allowed = cfg.get("allowed_decisions", allowed)
-                        break
-
-                # Формируем подсказку для пользователя
-                hints = []
-                if "approve" in allowed:
-                    hints.append("a = approve")
-                if "reject"  in allowed:
-                    hints.append("r = reject")
-                if "edit"    in allowed:
-                    hints.append("e = edit")
-
-                while True:
-                    choice = input(", ".join(hints) + ": ").strip().lower()
-
-                    if choice in ("a", "approve") and "approve" in allowed:
-                        decisions.append({"type": "approve"})
-                        break
-
-                    elif choice in ("r", "reject") and "reject" in allowed:
-                        message = input(
-                            "Сообщение для агента (причина отказа): "
-                        ).strip()
-                        decisions.append({
-                            "type": "reject",
-                            "message": message or "Нет причины",
-                        })
-                        break
-
-                    elif choice in ("e", "edit") and "edit" in allowed:
-                        # По желанию: редактирование аргументов перед выполнением
-                        print(f"Текущие аргументы: {args}")
-                        print("Введите изменения в формате key=value через запятую.")
-                        raw = input("Новые аргументы: ").strip()
-                        new_args = dict(args)
-                        for part in raw.split(","):
-                            part = part.strip()
-                            if "=" in part:
-                                k, v = part.split("=", 1)
-                                new_args[k.strip()] = v.strip()
-                        decisions.append({
-                            "type": "edit",
-                            "edited_action": {"name": name, "args": new_args},
-                        })
-                        break
-
-                    else:
-                        print(
-                            f"Недопустимый выбор. Варианты: {', '.join(hints)}"
-                        )
-
-            # Возобновляем агента с принятыми решениями
             result = agent.invoke(
                 Command(resume={"decisions": decisions}),
                 config=config,
             )
 
-        # ── Финальный ответ ───────────────────────────────────────────────
+        # После завершения выводим ответ агента
-        # Сообщения — объекты LangChain (AIMessage и др.),
+        if "messages" in result and result["messages"]:
-        # текст хранится в атрибуте .content, а не в ключе словаря.
+            last_msg = result["messages"][-1]
-        messages = result.get("messages", [])
+            print(f"\nАгент: {last_msg.get('content', '')}")
-        if messages:
+        else:
-            print(f"\nАгент: {messages[-1].content}")
+            print("\nАгент не вернул ответа.")
-
 
 if __name__ == "__main__":
     main()