Практическое задание №3: Память и подтверждение действий: solution.py

solutions/69a474cdc46fd26feae69896_Практическое_задание__3__Память_и_подтве/solution.py

@@ -0,0 +1,102 @@
+# solution.py
+
+import sys
+from typing import Any, Dict, Iterable, Tuple
+
+# 1️⃣ Rich для красивого вывода
+from rich.console import Console
+
+console = Console()
+
+# 2️⃣ LangGraph и необходимые компоненты
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import StateGraph
+from langgraph.prebuilt import create_chat_agent
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
+
+# 3️⃣ Определяем простой инструмент (пример)
+def echo_tool(args: Dict[str, Any]) -> str:
+    """Простейший инструмент, который возвращает переданный текст."""
+    return f"Эхо: {args.get('text', '')}"
+
+# 4️⃣ Создаём LLM
+llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
+
+# 5️⃣ Настраиваем память и агент с паузой перед инструментом
+memory = MemorySaver()
+
+agent = create_chat_agent(
+    llm=llm,
+    tools=[echo_tool],
+    system_prompt="Ты — полезный ассистент, который сначала спрашивает подтверждение перед вызовом инструмента.",
+    checkpointer=memory,
+    interrupt_before=["tools"],  # пауза перед каждым инструментом
+)
+
+# 6️⃣ Функция для обработки потокового ответа и подтверждения
+def ask_and_run(user_input: Dict[str, Any], config: Dict[str, Any]) -> None:
+    """
+    Отправляем запрос агенту, обрабатываем потоковые чанки,
+    при необходимости запрашиваем у пользователя подтверждение перед вызовом инструмента.
+    """
+    # Запускаем поток
+    for chunk in agent.stream(user_input, config=config, stream_mode=["messages", "updates"]):
+        state = agent.get_state(config)
+        chunk_type, chunk_data = chunk
+
+        # 6.1️⃣ Печатаем токены LLM в режиме потока
+        if chunk_type == "messages":
+            for msg in chunk_data:
+                if isinstance(msg, AIMessage):
+                    console.print(f"[bold cyan]AI:[/bold cyan] {msg.content}", end="")
+                elif isinstance(msg, ToolMessage):
+                    # Инструмент уже вызван – выводим результат
+                    console.print(f"\n[green]Tool result:[/green] {msg.content}")
+
+        # 6.2️⃣ Обрабатываем обновления (например, вызовы инструментов)
+        if chunk_type == "updates":
+            for update in chunk_data:
+                if isinstance(update, dict) and "tool_calls" in update:
+                    console.print("\n[magenta]Инструмент готов к выполнению:[/magenta]")
+                    tool_call = update["tool_calls"][0]
+                    name = tool_call["name"]
+                    args = tool_call.get("args", {})
+                    console.print(f"[yellow]{name}({args})[/yellow]")
+
+        # 6.3️⃣ Проверяем паузу перед инструментом
+        if "__interrupt__" in chunk_data and state.next == ("tools",):
+            # Получаем последний вызов инструмента
+            last_msg = state.values["messages"][-1]
+            tool_call = last_msg.tool_calls[0]
+            name = tool_call["name"]
+            args = tool_call.get("args", {})
+            console.print("\n\n[bold red]Пауза:[/bold red] Агент хочет вызвать инструмент.")
+            console.print(f"[yellow]{name}({args})[/yellow]")
+            answer = input("[green]Разрешить? (Y/n): [/green]").strip().lower()
+            if answer in ("", "y", "yes"):
+                # Продолжаем с того же состояния
+                ask_and_run(None, config)
+            else:
+                console.print("[red]Отменено. Конец диалога.[/red]")
+                sys.exit(0)
+
+# 7️⃣ Основной цикл чата
+def main() -> None:
+    thread_id = "thread-1"
+    config = {"configurable": {"thread_id": thread_id}}
+
+    console.print("[bold blue]Запуск агента. Введите 'exit' для завершения.[/bold blue]")
+    while True:
+        user_input = input("\n[cyan]Вы:[/cyan] ")
+        if user_input.lower() == "exit":
+            console.print("[bold green]До свидания![/bold green]")
+            break
+
+        ask_and_run(
+            {"messages": [{"role": "human", "content": user_input}]},
+            config,
+        )
+
+if __name__ == "__main__":
+    main()