Human-in-the-Loop через middleware: agent.py
This commit is contained in:
@@ -1,136 +1,100 @@
|
||||
import os
|
||||
import logging
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from langchain_ollama import OllamaEmbeddings, ChatOllama
|
||||
from langchain_qdrant import QdrantStore
|
||||
from langchain_core.documents import Document
|
||||
from langchain_core.runnables import RunnableLambda
|
||||
# ──────────────────────── Imports from the required stack ────────────────────────
|
||||
from langchain_ollama import ChatOllama
|
||||
from langchain.agents import create_agent
|
||||
from langchain.agents.middleware import HumanInTheLoopMiddleware
|
||||
from langgraph.checkpoint.memory import MemorySaver
|
||||
|
||||
# Настройка логирования
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
# ──────────────────────── LLM and tool definition ─────────────────────────────
|
||||
llm = ChatOllama(model="llama3")
|
||||
|
||||
def get_weather(city: str, date: str) -> str:
|
||||
"""Mock implementation of a weather‑lookup tool."""
|
||||
return f"Погода в {city} на {date}: солнечно, +25°C"
|
||||
|
||||
# ──────────────────────── Agent creation with HumanInTheLoopMiddleware ───────
|
||||
memory = MemorySaver()
|
||||
|
||||
agent = create_agent(
|
||||
model=llm,
|
||||
tools=[get_weather],
|
||||
system_prompt="Ты полезный ассистент",
|
||||
middleware=[
|
||||
HumanInTheLoopMiddleware(
|
||||
interrupt_on={
|
||||
"get_weather": True, # allow approve / edit / reject
|
||||
# "get_weather": {"allowed_decisions": ["approve", "reject"]}, # without edit
|
||||
},
|
||||
description_prefix="Подтвердите вызов инструмента",
|
||||
),
|
||||
],
|
||||
checkpointer=memory,
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Конфигурация подключения к Qdrant
|
||||
QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333")
|
||||
COLLECTION_NAME = os.getenv("QDRANT_COLLECTION", "knowledge_base")
|
||||
# ──────────────────────── Helper for the human‑in‑the‑loop loop ───────────────
|
||||
def run_agent_with_human_loop(user_message: str, thread_id: str = "session-1") -> None:
|
||||
"""
|
||||
Запускает агента с возможностью подтверждения вызова инструментов.
|
||||
user_message – сообщение пользователя (строка).
|
||||
thread_id – идентификатор сессии для сохранения истории.
|
||||
"""
|
||||
from langgraph.types import Command
|
||||
|
||||
# Инициализация эмбеддингов и модели Ollama
|
||||
try:
|
||||
embeddings = OllamaEmbeddings(model="nomic-embed-text")
|
||||
except Exception as e:
|
||||
logger.exception("Не удалось подключиться к Ollama Embeddings: %s", e)
|
||||
raise
|
||||
config = {"configurable": {"thread_id": thread_id}}
|
||||
|
||||
try:
|
||||
llm = ChatOllama(model="llama3")
|
||||
except Exception as e:
|
||||
logger.exception("Не удалось подключиться к Ollama LLM: %s", e)
|
||||
raise
|
||||
|
||||
# Инициализация хранилища Qdrant
|
||||
try:
|
||||
qdrant_store = QdrantStore(
|
||||
url=QDRANT_URL,
|
||||
collection_name=COLLECTION_NAME,
|
||||
embeddings=embeddings,
|
||||
# первый запуск
|
||||
result = agent.invoke(
|
||||
{"messages": [{"role": "human", "content": user_message}]},
|
||||
config=config,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.exception("Не удалось подключиться к Qdrant: %s", e)
|
||||
raise
|
||||
|
||||
# цикл подтверждений
|
||||
while "__interrupt__" in result:
|
||||
interrupt_value = result["__interrupt__"][0].value
|
||||
action_requests = interrupt_value["action_requests"]
|
||||
review_configs = interrupt_value["review_configs"]
|
||||
|
||||
decisions: List[Dict[str, Any]] = []
|
||||
|
||||
print("\n--- Подтверждение ---")
|
||||
for idx, (req, cfg) in enumerate(zip(action_requests, review_configs), start=1):
|
||||
name = req.get("name")
|
||||
args = req.get("args", {})
|
||||
description = req.get("description")
|
||||
|
||||
print(f"\n{idx}. Инструмент: {name}")
|
||||
print(f" Аргументы: {args}")
|
||||
if description:
|
||||
print(f" Описание: {description}")
|
||||
|
||||
# определяем доступные решения
|
||||
allowed = cfg.get("allowed_decisions", ["approve", "edit", "reject"])
|
||||
options = ", ".join(
|
||||
[f"{'a'=approve}" if d == "approve" else f"{'r'=reject}"
|
||||
for d in allowed]
|
||||
)
|
||||
# простейший ввод (без проверки)
|
||||
choice = input(f"a = approve, r = reject: ").strip().lower()
|
||||
if choice == "a":
|
||||
decisions.append({"type": "approve"})
|
||||
elif choice == "r":
|
||||
msg = input("Сообщение для агента (причина отказа): ")
|
||||
decisions.append({"type": "reject", "message": msg})
|
||||
else:
|
||||
# если пользователь ввёл что‑то другое, считаем reject
|
||||
decisions.append({"type": "reject", "message": "неизвестное решение"})
|
||||
|
||||
# возобновляем выполнение
|
||||
result = agent.invoke(Command(resume={"decisions": decisions}), config=config)
|
||||
|
||||
# вывод финального ответа
|
||||
final_message = result["messages"][-1]["content"]
|
||||
print("\nАгент:", final_message)
|
||||
|
||||
|
||||
# ==========================
|
||||
# Middleware для обработки ошибок и логирования
|
||||
# ==========================
|
||||
def middleware(func):
|
||||
"""
|
||||
Декоратор, который оборачивает функцию в try/except,
|
||||
логирует входные параметры и результат.
|
||||
"""
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
func_name = func.__name__
|
||||
logger.debug("Вызов %s с args=%s kwargs=%s", func_name, args, kwargs)
|
||||
try:
|
||||
result = func(*args, **kwargs)
|
||||
logger.debug("Результат %s: %s", func_name, result)
|
||||
return result
|
||||
except Exception as exc:
|
||||
logger.exception("Ошибка в %s: %s", func_name, exc)
|
||||
raise
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
# ==========================
|
||||
# RAG инструменты
|
||||
# ==========================
|
||||
@middleware
|
||||
def search_knowledge_base(query: str) -> List[Document]:
|
||||
"""
|
||||
Поиск релевантных документов в Qdrant.
|
||||
Возвращает список Document с полями content и metadata.
|
||||
"""
|
||||
if not query:
|
||||
raise ValueError("Запрос не может быть пустым")
|
||||
|
||||
try:
|
||||
docs = qdrant_store.similarity_search(query, k=5)
|
||||
logger.info("Найдено %d документов по запросу '%s'", len(docs), query)
|
||||
return docs
|
||||
except Exception as exc:
|
||||
logger.exception("Ошибка при поиске: %s", exc)
|
||||
raise
|
||||
|
||||
|
||||
@middleware
|
||||
def add_to_knowledge_base(content: str, metadata: Dict[str, Any] | None = None) -> None:
|
||||
"""
|
||||
Добавляет новый документ в Qdrant.
|
||||
content – текст документа,
|
||||
metadata – произвольные метаданные (может быть None).
|
||||
"""
|
||||
if not content:
|
||||
raise ValueError("Содержимое не может быть пустым")
|
||||
|
||||
try:
|
||||
doc = Document(page_content=content, metadata=metadata or {})
|
||||
qdrant_store.add_documents([doc])
|
||||
logger.info("Документ добавлен в коллекцию '%s'", COLLECTION_NAME)
|
||||
except Exception as exc:
|
||||
logger.exception("Ошибка при добавлении документа: %s", exc)
|
||||
raise
|
||||
|
||||
|
||||
# ==========================
|
||||
# Пример использования (можно использовать как тесты)
|
||||
# ==========================
|
||||
# ──────────────────────── Пример использования ───────────────────────────────
|
||||
if __name__ == "__main__":
|
||||
# Минимальный набор unit‑тестов
|
||||
import unittest
|
||||
|
||||
class TestAgent(unittest.TestCase):
|
||||
def test_add_and_search(self):
|
||||
test_content = "Тестовый документ для проверки RAG."
|
||||
test_metadata = {"source": "unit_test"}
|
||||
|
||||
# Добавляем документ
|
||||
add_to_knowledge_base(test_content, test_metadata)
|
||||
|
||||
# Ищем по ключевому слову
|
||||
results = search_knowledge_base("тестовый")
|
||||
self.assertTrue(any(test_content in doc.page_content for doc in results))
|
||||
|
||||
def test_empty_query(self):
|
||||
with self.assertRaises(ValueError):
|
||||
search_knowledge_base("")
|
||||
|
||||
def test_empty_content(self):
|
||||
with self.assertRaises(ValueError):
|
||||
add_to_knowledge_base("")
|
||||
|
||||
unittest.main(argv=["first-arg-is-ignored"], exit=False)
|
||||
user_input = input("Вы: ")
|
||||
run_agent_with_human_loop(user_input)
|
||||
Reference in New Issue
Block a user