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

2026-05-27 10:34:07 +00:00
parent 6531446386
commit ce36785e56
1 changed files with 100 additions and 0 deletions
@@ -0,0 +1,100 @@
 # solution.py
 """
 Human-in-the-Loop (HIL) через middleware на основе FastMCP.
 В этом примере реализован простой сервер, который имеет один эндпоинт /process.
 Перед выполнением запроса к этому эндпоинту пользователь должен подтвердить действие,
 что демонстрирует принцип HIL: человек вмешивается в поток выполнения.
 Требования:
 - Используется только стек FastMCP (не FastAPI).
 - Middleware реализует запрос подтверждения через консольный ввод.
 - Если пользователь отвечает "yes" (любое регистровое сочетание), запрос
  продолжается, иначе возвращается сообщение об отмене.
 """
 import sys
 from fastmcp import FastMCP, Request, Response
 # --------------------------------------------------------------------
 # Middleware для Human-in-the-Loop
 # --------------------------------------------------------------------
 class HILMiddleware:
    """
    Middleware, который перехватывает запросы к /process и запрашивает подтверждение у пользователя.
    """
    def __init__(self, app: FastMCP):
        self.app = app
    async def __call__(self, request: Request) -> Response:
        # Проверяем путь запроса
        if request.path == "/process":
            # Запрос подтверждения
            print("\n=== Human-in-the-Loop ===")
            print(f"Запрос к {request.method} {request.path}")
            confirm = input("Вы действительно хотите выполнить это действие? (yes/no): ").strip().lower()
            if confirm != "yes":
                return Response(
                    status_code=403,
                    body="Действие отменено пользователем.",
                    headers={"Content-Type": "text/plain"},
                )
        # Если подтверждение получено или путь не /process, продолжаем цепочку
        return await self.app(request)
 # --------------------------------------------------------------------
 # Основное приложение FastMCP
 # --------------------------------------------------------------------
 app = FastMCP()
@app.route("/process", methods=["POST"])
 async def process_handler(request: Request) -> Response:
    """
    Пример обработчика, который выполняет некоторую задачу.
    В реальном приложении здесь могла бы быть бизнес-логика.
    """
    # Здесь можно добавить любую логику
    return Response(
        status_code=200,
        body="Действие выполнено успешно.",
        headers={"Content-Type": "text/plain"},
    )
@app.route("/", methods=["GET"])
 async def index(request: Request) -> Response:
    """
    Статическая страница приветствия.
    """
    html = (
        "<html>"
        "<head><title>HIL Middleware Demo</title></head>"
        "<body>"
        "<h1>Добро пожаловать!</h1>"
        "<p>Отправьте POST запрос на <code>/process</code>, чтобы увидеть HIL.</p>"
        "</body>"
        "</html>"
    )
    return Response(
        status_code=200,
        body=html,
        headers={"Content-Type": "text/html"},
    )
 # Оборачиваем приложение в middleware
 app = HILMiddleware(app)
 # --------------------------------------------------------------------
 # Запуск сервера
 # --------------------------------------------------------------------
 if __name__ == "__main__":
    # FastMCP использует uvicorn под капотом, но запускаем напрямую
    import uvicorn
    uvicorn.run(
        "solution:app",
        host="127.0.0.1",
        port=8000,
        log_level="info",
        reload=False,
    )