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

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

@@ -1,157 +1,86 @@
-#!/usr/bin/env python3
+# solution.py
-# -*- coding: utf-8 -*-
 
 """
-solution.py
+Пример агента с Human-in-the-Loop через middleware.
-
+Запускается из командной строки и демонстрирует работу интерактивного подтверждения вызова инструмента.
-Пример использования HumanInTheLoopMiddleware в LangChain Agent.
-При каждом вызове инструмента агент останавливается и запрашивает у пользователя подтверждение
-(approve / reject). После принятия решения выполнение возобновляется через Command.
-
-Требования:
-    pip install langchain langgraph openai rich
 """
 
-import json
+from langgraph.prebuilt import create_react_agent
-from typing import Any, Dict, List
-
-# --------------------------------------------------------------------------- #
-# 1. Подключаем необходимые модули LangChain и LangGraph
-# --------------------------------------------------------------------------- #
-
-from langchain.agents import create_agent
-from langchain.agents.middleware import HumanInTheLoopMiddleware
-from langgraph.checkpoint.memory import MemorySaver
-from langgraph.types import Command
 from langchain.tools import tool
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.types import Command, StateGraph
+from langgraph.moderation import HumanInTheLoopMiddleware
 
-# --------------------------------------------------------------------------- #
+# ------------------------------------------------------------
-# 2. Определяем простой инструмент get_weather
+# 1. Определяем простой инструмент
-# --------------------------------------------------------------------------- #
+# ------------------------------------------------------------
-
+@tool
-@tool("get_weather", "Получить погоду в указанном городе и дате")
 def get_weather(city: str, date: str) -> str:
-    """
+    """Возвращает погоду в городе на указанную дату."""
-    Возвращает фиктивную информацию о погоде.
+    return f"Погода в {city} на {date}: солнечно 25°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
-# --------------------------------------------------------------------------- #
 
+# ------------------------------------------------------------
+# 2. Создаём агент с HumanInTheLoopMiddleware
+# ------------------------------------------------------------
 memory = MemorySaver()
 
-agent = create_agent(
+agent = create_react_agent(
-    model=llm,
     tools=[get_weather],
-    system_prompt="Ты полезный ассистент, помогающий пользователю.",
+    system_prompt="Ты полезный ассистент, отвечай точно.",
     middleware=[
         HumanInTheLoopMiddleware(
-            interrupt_on={
+            interrupt_on={"get_weather": True},  # прерываем только при вызове get_weather
-                "get_weather": True,  # разрешаем все решения
-            },
             description_prefix="Подтвердите вызов инструмента",
         ),
     ],
     checkpointer=memory,
 )
 
-# --------------------------------------------------------------------------- #
+# ------------------------------------------------------------
-# 5. Функция для получения решений от пользователя
+# 3. Запускаем агент и обрабатываем интерактивный цикл
-# --------------------------------------------------------------------------- #
+# ------------------------------------------------------------
-
+def run_agent():
-def ask_decisions(action_requests: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     """
-    Для каждого запроса к инструменту выводим информацию и запрашиваем у пользователя решение.
+    Вводим сообщение от пользователя, запускаем агента,
-    Возвращаем список словарей с решениями в том же порядке, что и action_requests.
+    обрабатываем прерывания и продолжаем работу до завершения.
     """
-    decisions = []
+    # Инициализируем конфиг с thread_id
-    print("\n--- Подтверждение вызова инструмента ---")
+    config = {"configurable": {"thread_id": "session-1"}}
-    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:
+    user_input = input("Введите запрос: ")
-            print(f"   Описание: {description}")
-        print(f"   Аргументы: {json.dumps(args, ensure_ascii=False)}")
 
-        while True:
+    # Первый вызов агента
-            choice = input("a = approve, r = reject: ").strip().lower()
+    result = agent.invoke(
-            if choice == "a":
+        {"messages": [{"role": "human", "content": user_input}]},
-                decisions.append({"type": "approve"})
+        config=config,
-                break
+    )
-            elif choice == "r":
+
-                msg = input(
+    # Цикл обработки прерываний
-                    "Введите причину отказа (можно оставить пустой): "
+    while "__interrupt__" in result:
-                ).strip()
+        interrupt_payload = result["__interrupt__"][0].value
-                decisions.append({"type": "reject", "message": msg})
+        action_requests = interrupt_payload.get("action_requests", [])
-                break
+
+        decisions = []
+        for req in action_requests:
+            print("\nИнструмент:", req["name"])
+            print("Аргументы:", req["args"])
+
+            # Запрашиваем решение пользователя
+            decision = input("a=approve, r=reject: ").strip().lower()
+            if decision == "r":
+                reason = input("Причина отказа: ").strip()
+                decisions.append({"type": "reject", "message": reason})
             else:
-                print("Неверный ввод. Пожалуйста, введите 'a' или 'r'.")
+                decisions.append({"type": "approve"})
-    return decisions
 
-# --------------------------------------------------------------------------- #
+        # Возобновляем работу агента
-# 6. Основной цикл взаимодействия с агентом
+        result = agent.invoke(Command(resume={"decisions": decisions}), config=config)
-# --------------------------------------------------------------------------- #
 
-def main() -> None:
+    # Выводим финальный ответ агента
-    """
+    final_message = result["messages"][-1]["content"]
-    Запускает чат-цикл: пользователь вводит сообщение,
+    print("\nОтвет агента:", final_message)
-    агент обрабатывает его, при необходимости запрашивает подтверждение.
-    После завершения выводится финальный ответ агента.
-    """
-    thread_id = "session-1"
-    config = {"configurable": {"thread_id": thread_id}}
 
-    print("Привет! Я ассистент. Введите ваш запрос (или 'выход' для завершения).")
-
-    while True:
-        user_msg = input("\nВы: ").strip()
-        if user_msg.lower() in ("выход", "quit", "exit"):
-            print("До свидания!")
-            break
-
-        # Первый вызов агента
-        result = agent.invoke(
-            {"messages": [{"role": "human", "content": user_msg}]},
-            config=config,
-        )
-
-        # Цикл подтверждений
-        while "__interrupt__" in result:
-            interrupt_value = result["__interrupt__"][0].value
-            action_requests = interrupt_value.get("action_requests", [])
-            # review_configs не используется в этом примере, но можно вывести при желании
-
-            decisions = ask_decisions(action_requests)
-
-            # Возобновляем выполнение агента с решениями
-            result = agent.invoke(
-                Command(resume={"decisions": decisions}),
-                config=config,
-            )
-
-        # После завершения выводим ответ агента
-        if "messages" in result and result["messages"]:
-            last_msg = result["messages"][-1]
-            print(f"\nАгент: {last_msg.get('content', '')}")
-        else:
-            print("\nАгент не вернул ответа.")
 
 if __name__ == "__main__":
-    main()
+    run_agent()