Обновить solution.py
This commit is contained in:
+101
-98
@@ -1,108 +1,111 @@
|
||||
<|channel|>final code<|message|>import time
|
||||
from typing import Callable, Awaitable
|
||||
# solution.py
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Агент с HumanInTheLoopMiddleware: при каждом вызове инструмента
|
||||
агент останавливается, пользователь подтверждает (approve/reject),
|
||||
после чего выполнение возобновляется через Command.
|
||||
"""
|
||||
|
||||
import uvicorn
|
||||
from fastapi import FastAPI, Request, Response, HTTPException
|
||||
from fastapi.middleware.base import BaseHTTPMiddleware
|
||||
from langchain.llms.openai import OpenAI
|
||||
from qdrant_client import QdrantClient
|
||||
from rich.console import Console
|
||||
import json
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langchain_core.tools import tool
|
||||
from langchain.agents import create_agent
|
||||
from langchain.agents.middleware import HumanInTheLoopMiddleware
|
||||
from langgraph.checkpoint.memory import MemorySaver
|
||||
from langgraph.types import Command
|
||||
|
||||
# Консоль для красивого логирования
|
||||
console = Console()
|
||||
# 1. Инструмент
|
||||
@tool
|
||||
def get_weather(city: str, date: str = "сегодня") -> str:
|
||||
"""Получить погоду в городе на указанную дату."""
|
||||
return f"В городе {city} на {date}: солнечно, 25°C."
|
||||
|
||||
# Инициализация клиента Qdrant (предполагается, что он уже запущен)
|
||||
qdrant = QdrantClient(host="localhost", port=6333)
|
||||
# 2. LLM
|
||||
llm = ChatOpenAI(
|
||||
model="gpt-4o-mini",
|
||||
temperature=0,
|
||||
)
|
||||
|
||||
# Инициализация LLM (замените на свой ключ API)
|
||||
llm = OpenAI(api_key="YOUR_OPENAI_API_KEY")
|
||||
# 3. Память
|
||||
memory = MemorySaver()
|
||||
|
||||
app = FastAPI(title="Human‑in‑the‑loop Middleware Demo")
|
||||
|
||||
|
||||
class HumanInLoopMiddleware(BaseHTTPMiddleware):
|
||||
"""
|
||||
Middleware, который позволяет прерывать и возобновлять обработку запросов.
|
||||
Если в заголовке `X-HITL-Interrupt` присутствует значение 'true',
|
||||
запрос будет поставлен в очередь на паузу. При получении
|
||||
запроса с заголовком `X-HITL-Resume` выполнение продолжается.
|
||||
"""
|
||||
|
||||
def __init__(self, app: FastAPI):
|
||||
super().__init__(app)
|
||||
# Очередь для хранения приостановленных запросов (id -> request data)
|
||||
self.paused_requests = {}
|
||||
|
||||
async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
|
||||
interrupt_flag = request.headers.get("X-HITL-Interrupt", "false").lower()
|
||||
resume_flag = request.headers.get("X-HITL-Resume", "false").lower()
|
||||
|
||||
# Если запрос должен быть приостановлен
|
||||
if interrupt_flag == "true":
|
||||
req_id = str(time.time())
|
||||
self.paused_requests[req_id] = {
|
||||
"method": request.method,
|
||||
"url": request.url.path,
|
||||
"body": await request.body(),
|
||||
"headers": dict(request.headers),
|
||||
}
|
||||
console.log(f"[yellow]Request {req_id} paused[/]")
|
||||
return Response(content=f"Request paused with id: {req_id}", status_code=202)
|
||||
|
||||
# Если запрос должен возобновиться
|
||||
if resume_flag == "true":
|
||||
req_id = request.query_params.get("resume_id")
|
||||
if not req_id or req_id not in self.paused_requests:
|
||||
raise HTTPException(status_code=400, detail="Invalid or missing resume_id")
|
||||
|
||||
paused = self.paused_requests.pop(req_id)
|
||||
console.log(f"[green]Resuming request {req_id}[/]")
|
||||
# Создаём новый Request объект из сохранённых данных
|
||||
new_request = Request(
|
||||
scope={
|
||||
"type": "http",
|
||||
"method": paused["method"],
|
||||
"path": paused["url"],
|
||||
"headers": [(k.encode(), v.encode()) for k, v in paused["headers"].items()],
|
||||
"query_string": b"",
|
||||
"client": request.client,
|
||||
"server": request.scope.get("server"),
|
||||
# 4. Агент с HumanInTheLoopMiddleware
|
||||
agent = create_agent(
|
||||
model=llm,
|
||||
tools=[get_weather],
|
||||
system_prompt="Ты полезный ассистент.",
|
||||
middleware=[
|
||||
HumanInTheLoopMiddleware(
|
||||
interrupt_on={
|
||||
"get_weather": True,
|
||||
},
|
||||
receive=lambda: {"type": "http.request", "body": paused["body"]},
|
||||
)
|
||||
return await call_next(new_request)
|
||||
|
||||
# Нормальная обработка
|
||||
response = await call_next(request)
|
||||
return response
|
||||
|
||||
|
||||
app.add_middleware(HumanInLoopMiddleware)
|
||||
|
||||
|
||||
@app.post("/process")
|
||||
async def process_endpoint(data: dict):
|
||||
"""
|
||||
Пример эндпоинта, который использует LLM и Qdrant.
|
||||
"""
|
||||
console.log("[blue]Processing request...[/]")
|
||||
# Сохраняем запрос в Qdrant
|
||||
qdrant.upsert(
|
||||
collection_name="requests",
|
||||
points=[
|
||||
{
|
||||
"id": str(time.time()),
|
||||
"vector": [0.1, 0.2], # placeholder vector
|
||||
"payload": data,
|
||||
}
|
||||
description_prefix="Подтвердите вызов инструмента",
|
||||
),
|
||||
],
|
||||
)
|
||||
# Генерируем ответ через LLM
|
||||
prompt = f"User asked: {data.get('question', '')}"
|
||||
answer = llm(prompt)
|
||||
console.log("[blue]LLM response generated[/]")
|
||||
return {"answer": answer}
|
||||
checkpointer=memory,
|
||||
)
|
||||
|
||||
# 5. Сбор решений от пользователя
|
||||
def get_user_decisions(action_requests: list[dict], review_configs: list[dict]) -> list[dict]:
|
||||
decisions = []
|
||||
for action, review_cfg in zip(action_requests, review_configs):
|
||||
name = action.get("name", "unknown")
|
||||
args = action.get("args", {})
|
||||
description = action.get("description", "")
|
||||
allowed = review_cfg.get("allowed_decisions", ["approve", "reject"])
|
||||
|
||||
print(f"\n--- Подтверждение ---")
|
||||
print(f"Инструмент: {name}")
|
||||
print(f"Аргументы: {json.dumps(args, ensure_ascii=False)}")
|
||||
if description:
|
||||
print(f"Описание: {description}")
|
||||
print(f"Разрешённые решения: {', '.join(allowed)}")
|
||||
|
||||
while True:
|
||||
choice = input("a=approve, r=reject: ").strip().lower()
|
||||
if choice in ("a", "approve"):
|
||||
decisions.append({"type": "approve"})
|
||||
break
|
||||
elif choice in ("r", "reject"):
|
||||
msg = input("Причина отказа: ").strip()
|
||||
decisions.append({
|
||||
"type": "reject",
|
||||
"message": msg or "Запрос отклонён пользователем",
|
||||
})
|
||||
break
|
||||
else:
|
||||
print("Неверный ввод. Введите 'a' или 'r'.")
|
||||
return decisions
|
||||
|
||||
# 6. Основной цикл
|
||||
def main() -> None:
|
||||
config = {"configurable": {"thread_id": "session-1"}}
|
||||
print("Привет! Введите запрос или 'выход' для завершения.")
|
||||
|
||||
while True:
|
||||
user_input = input("\nВы: ").strip()
|
||||
if user_input.lower() in {"выход", "exit", "quit"}:
|
||||
print("Завершение работы.")
|
||||
break
|
||||
|
||||
result = agent.invoke(
|
||||
{"messages": [{"role": "human", "content": user_input}]},
|
||||
config=config,
|
||||
)
|
||||
|
||||
while "__interrupt__" in result:
|
||||
interrupt_value = result["__interrupt__"][0].value
|
||||
action_requests = interrupt_value.get("action_requests", [])
|
||||
review_configs = interrupt_value.get("review_configs", [])
|
||||
|
||||
decisions = get_user_decisions(action_requests, review_configs)
|
||||
|
||||
result = agent.invoke(
|
||||
Command(resume={"decisions": decisions}),
|
||||
config=config,
|
||||
)
|
||||
|
||||
print(f"\nАгент: {result['messages'][-1].content}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
main()
|
||||
Reference in New Issue
Block a user