Агент с RAG‑памятью: cli_client.py

solutions/6a02e23da6fe2e4ac16acf65_Агент_с_RAG_памятью/cli_client.py

@@ -4,7 +4,7 @@
 """
 cli_client.py
 
-Интерактивный клиент для работы с агентом RAG‑памяти.
+Интерактивный клиент для работы с агентом RAG.
 Поддерживает команды:
     /add <title> <content>   – добавить документ в базу знаний
     /search <query>          – выполнить семантический поиск
@@ -14,110 +14,122 @@ cli_client.py
 import sys
 from typing import List
 
-# Предполагается, что модули с инструментами уже реализованы и находятся в пакете `rag_agent`.
+# Rich для красивого вывода
-# Импортируем необходимые функции.
 try:
-    from rag_agent.tools import add_to_knowledge_base, search_knowledge_base  # noqa: F401
+    from rich.console import Console
-except Exception as exc:
+    from rich.table import Table
-    print(f"Ошибка при импорте инструментов: {exc}", file=sys.stderr)
+except ImportError:  # pragma: no cover
+    print("Необходимо установить пакет 'rich'")
     sys.exit(1)
 
+# Инструменты агента (предполагается, что они реализованы в другом модуле)
+try:
+    from rag_tools import search_knowledge_base, add_to_knowledge_base
+except Exception as exc:  # pragma: no cover
+    print(f"Ошибка при импорте инструментов: {exc}")
+    sys.exit(1)
 
-def parse_command(line: str) -> (str, List[str]):
+console = Console()
+
+
+def _parse_command(line: str) -> List[str]:
     """
-    Разбирает строку команды и возвращает имя команды и список аргументов.
+    Разбирает строку ввода пользователя.
-    Команда начинается с символа '/'.
+    Возвращает список токенов, разделённых пробелами,
+    но сохраняет аргументы в кавычках как один элемент.
     """
-    if not line.startswith("/"):
+    import shlex
-        raise ValueError("Команды должны начинаться со знака '/'.")
+
-    parts = line.strip().split(maxsplit=1)
+    try:
-    cmd = parts[0]
+        return shlex.split(line)
-    args_str = parts[1] if len(parts) > 1 else ""
+    except ValueError as exc:  # pragma: no cover
-    # Для /add аргументы разделены пробелом: title content
+        console.print(f"[red]Ошибка разбора команды:[/red] {exc}")
-    # Чтобы сохранить пробелы в содержимом, делаем split только один раз.
+        return []
-    return cmd, [args_str]
 
 
-def handle_add(args: List[str]) -> None:
+def _handle_add(tokens: List[str]) -> None:
     """
     Обработчик команды /add.
-    Ожидает строку вида "<title> <content>".
+    Ожидает два аргумента: title и content (в кавычках).
     """
-    if not args or not args[0]:
+    if len(tokens) < 3:
-        print("Ошибка: нужно указать заголовок и содержимое.")
+        console.print("[red]Неверное количество аргументов. Используйте:/add \"title\" \"content\"[/red]")
         return
-    # Разделяем на заголовок и контент. Заголовок – первое слово, остальное – контент.
+    title = tokens[1]
-    parts = args[0].split(maxsplit=1)
+    content = " ".join(tokens[2:])
-    if len(parts) < 2:
-        print("Ошибка: нужно указать как минимум заголовок и содержимое.")
-        return
-    title, content = parts[0], parts[1]
     try:
         add_to_knowledge_base(content=content, title=title)
-        print(f"Документ '{title}' успешно добавлен в базу знаний.")
+        console.print(f"[green]Документ '{title}' успешно добавлен.[/green]")
-    except Exception as exc:
+    except Exception as exc:  # pragma: no cover
-        print(f"Ошибка при добавлении документа: {exc}")
+        console.print(f"[red]Ошибка при добавлении:[/red] {exc}")
 
 
-def handle_search(args: List[str]) -> None:
+def _handle_search(tokens: List[str]) -> None:
     """
     Обработчик команды /search.
-    Ожидает строку запроса.
+    Ожидает один аргумент – запрос (в кавычках).
     """
-    if not args or not args[0]:
+    if len(tokens) < 2:
-        print("Ошибка: нужно указать поисковый запрос.")
+        console.print("[red]Неверное количество аргументов. Используйте:/search \"query\"[/red]")
         return
-    query = args[0]
+    query = " ".join(tokens[1:])
     try:
         results = search_knowledge_base(query=query, max_results=5)
         if not results:
-            print("Ничего не найдено.")
+            console.print("[yellow]Результаты не найдены.[/yellow]")
             return
-        print(f"Найдено {len(results)} результатов:")
+
-        for i, res in enumerate(results, 1):
+        table = Table(title="Поиск", show_lines=True)
-            # Предполагаем, что результат – словарь с полями 'title' и 'content'.
+        table.add_column("Номер", style="cyan")
-            title = res.get("metadata", {}).get("title", "Без названия")
+        table.add_column("Заголовок", style="magenta")
-            snippet = res.get("content")[:200] + ("..." if len(res.get("content", "")) > 200 else "")
+        table.add_column("Содержание")
-            print(f"\n{i}. {title}\n   {snippet}")
+
-    except Exception as exc:
+        for idx, res in enumerate(results, start=1):
-        print(f"Ошибка при поиске: {exc}")
+            # Предполагается, что результат – словарь с ключами title и content
+            title = res.get("title", "Без заголовка")
+            content = res.get("content", "")
+            snippet = (content[:200] + "...") if len(content) > 200 else content
+            table.add_row(str(idx), title, snippet)
+
+        console.print(table)
+    except Exception as exc:  # pragma: no cover
+        console.print(f"[red]Ошибка при поиске:[/red] {exc}")
 
 
 def main() -> None:
     """
-    Основной цикл интерактивного клиента.
+    Основной цикл CLI.
     """
-    print("=== RAG‑память CLI ===")
+    console.print("[bold green]Добро пожаловать в RAG-агент![/bold green]")
-    print("Доступные команды:")
+    console.print("Доступные команды:")
-    print("/add <title> <content>   – добавить документ в базу знаний")
+    console.print("/add \"title\" \"content\"   – добавить документ")
-    print("/search <query>          – выполнить семантический поиск")
+    console.print("/search \"query\"           – выполнить поиск")
-    print("/quit                    – выйти из программы\n")
+    console.print("/quit                       – выйти из программы")
 
     while True:
         try:
-            line = input("> ").strip()
+            line = console.input("\n[bold blue]Введите команду[/bold blue]> ")
-        except (EOFError, KeyboardInterrupt):
+        except (KeyboardInterrupt, EOFError):  # pragma: no cover
-            print("\nВыход.")
+            console.print("\n[bold red]Выход...[/bold red]")
             break
 
-        if not line:
+        if not line.strip():
             continue
 
-        try:
+        tokens = _parse_command(line)
-            cmd, args = parse_command(line)
+        if not tokens:
-        except ValueError as ve:
-            print(ve)
             continue
 
-        if cmd == "/add":
+        command = tokens[0].lower()
-            handle_add(args)
+        if command == "/add":
-        elif cmd == "/search":
+            _handle_add(tokens)
-            handle_search(args)
+        elif command == "/search":
-        elif cmd == "/quit":
+            _handle_search(tokens)
-            print("До свидания!")
+        elif command in {"/quit", "exit"}:
+            console.print("[bold red]Выход...[/bold red]")
             break
         else:
-            print(f"Неизвестная команда: {cmd}")
+            console.print(f"[red]Неизвестная команда: {command}[/red]")
 
 
 if __name__ == "__main__":