diff --git a/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py b/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py index d9fbfc2..44d2743 100644 --- a/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py +++ b/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py @@ -1,157 +1,86 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- +# solution.py """ -solution.py - -Пример использования HumanInTheLoopMiddleware в LangChain Agent. -При каждом вызове инструмента агент останавливается и запрашивает у пользователя подтверждение -(approve / reject). После принятия решения выполнение возобновляется через Command. - -Требования: - pip install langchain langgraph openai rich +Пример агента с Human-in-the-Loop через middleware. +Запускается из командной строки и демонстрирует работу интерактивного подтверждения вызова инструмента. """ -import json -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 langgraph.prebuilt import create_react_agent 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 -# --------------------------------------------------------------------------- # - -@tool("get_weather", "Получить погоду в указанном городе и дате") +# ------------------------------------------------------------ +# 1. Определяем простой инструмент +# ------------------------------------------------------------ +@tool def get_weather(city: str, date: str) -> str: - """ - Возвращает фиктивную информацию о погоде. - В реальном проекте здесь можно подключиться к 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 -# --------------------------------------------------------------------------- # + """Возвращает погоду в городе на указанную дату.""" + return f"Погода в {city} на {date}: солнечно 25°C." +# ------------------------------------------------------------ +# 2. Создаём агент с HumanInTheLoopMiddleware +# ------------------------------------------------------------ memory = MemorySaver() -agent = create_agent( - model=llm, +agent = create_react_agent( tools=[get_weather], - system_prompt="Ты полезный ассистент, помогающий пользователю.", + system_prompt="Ты полезный ассистент, отвечай точно.", middleware=[ HumanInTheLoopMiddleware( - interrupt_on={ - "get_weather": True, # разрешаем все решения - }, + interrupt_on={"get_weather": True}, # прерываем только при вызове get_weather description_prefix="Подтвердите вызов инструмента", ), ], checkpointer=memory, ) -# --------------------------------------------------------------------------- # -# 5. Функция для получения решений от пользователя -# --------------------------------------------------------------------------- # - -def ask_decisions(action_requests: List[Dict[str, Any]]) -> List[Dict[str, Any]]: +# ------------------------------------------------------------ +# 3. Запускаем агент и обрабатываем интерактивный цикл +# ------------------------------------------------------------ +def run_agent(): """ - Для каждого запроса к инструменту выводим информацию и запрашиваем у пользователя решение. - Возвращаем список словарей с решениями в том же порядке, что и action_requests. + Вводим сообщение от пользователя, запускаем агента, + обрабатываем прерывания и продолжаем работу до завершения. """ - decisions = [] - print("\n--- Подтверждение вызова инструмента ---") - for idx, req in enumerate(action_requests): - name = req.get("name", "неизвестный инструмент") - args = req.get("args", {}) - description = req.get("description", "") + # Инициализируем конфиг с thread_id + config = {"configurable": {"thread_id": "session-1"}} - print(f"\n{idx + 1}. Инструмент: {name}") - if description: - print(f" Описание: {description}") - print(f" Аргументы: {json.dumps(args, ensure_ascii=False)}") + # Запрашиваем ввод от пользователя + user_input = input("Введите запрос: ") - 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 + # Первый вызов агента + result = agent.invoke( + {"messages": [{"role": "human", "content": user_input}]}, + config=config, + ) + + # Цикл обработки прерываний + while "__interrupt__" in result: + interrupt_payload = result["__interrupt__"][0].value + action_requests = interrupt_payload.get("action_requests", []) + + 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'.") - return decisions + decisions.append({"type": "approve"}) -# --------------------------------------------------------------------------- # -# 6. Основной цикл взаимодействия с агентом -# --------------------------------------------------------------------------- # + # Возобновляем работу агента + result = agent.invoke(Command(resume={"decisions": decisions}), config=config) -def main() -> None: - """ - Запускает чат-цикл: пользователь вводит сообщение, - агент обрабатывает его, при необходимости запрашивает подтверждение. - После завершения выводится финальный ответ агента. - """ - thread_id = "session-1" - config = {"configurable": {"thread_id": thread_id}} + # Выводим финальный ответ агента + final_message = result["messages"][-1]["content"] + print("\nОтвет агента:", final_message) - 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() \ No newline at end of file + run_agent() \ No newline at end of file