diff --git a/solutions/69a474cdc46fd26feae69896_Практическое_задание__3__Память_и_подтве/solution.py b/solutions/69a474cdc46fd26feae69896_Практическое_задание__3__Память_и_подтве/solution.py index d2d7018..fbb700f 100644 --- a/solutions/69a474cdc46fd26feae69896_Практическое_задание__3__Память_и_подтве/solution.py +++ b/solutions/69a474cdc46fd26feae69896_Практическое_задание__3__Память_и_подтве/solution.py @@ -1,128 +1,116 @@ # solution.py import os -from typing import Any, Dict, Iterable, Tuple +from typing import Any, Dict, Iterable, Tuple, Optional -from langchain_openai import ChatOpenAI +from langchain.chat_models import ChatOpenAI +from langchain.tools import BaseTool from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph -from langgraph.prebuilt.tool_executor import ToolExecutor +from langgraph.prebuilt import create_agent from rich.console import Console +# ------------------------------------------------------------ +# Конфигурация LLM и инструмента +# ------------------------------------------------------------ + +# Для работы нужен ключ OpenAI. Если его нет – создаём фиктивный LLM. +OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") +if OPENAI_API_KEY: + llm = ChatOpenAI(temperature=0, openai_api_key=OPENAI_API_KEY) +else: + # Минимальный заглушка‑LLM (не делает ничего полезного) + class DummyLLM: + def __call__(self, *args, **kwargs): + return {"content": "Dummy response"} + llm = DummyLLM() + +# Пример простого инструмента +class EchoTool(BaseTool): + name: str = "echo" + description: str = "Возвращает переданный текст без изменений." + def _run(self, text: str) -> str: + return f"Echo: {text}" + +my_tool = EchoTool() + +# ------------------------------------------------------------ +# Создание агента с памятью и паузой +# ------------------------------------------------------------ + +memory = MemorySaver() +agent = create_agent( + model=llm, + tools=[my_tool], + system_prompt="Ты полезный ассистент. Перед каждым вызовом инструмента спрашивай разрешение.", + checkpointer=memory, # нужна для сохранения состояния паузы + interrupt_before=["tools"], # остановка перед инструментом +) + console = Console() -# ---------- 1. Настройка модели и инструментов ---------- -llm = ChatOpenAI( - model="gpt-4o-mini", - temperature=0, -) +# ------------------------------------------------------------ +# Функция взаимодействия с пользователем +# ------------------------------------------------------------ -def echo_tool(args: Dict[str, Any]) -> str: - """Простейший инструмент для демонстрации.""" - return f"Эхо: {args.get('message', '')}" - -tools = [echo_tool] -tool_executor = ToolExecutor(tools) - - -# ---------- 2. Создание графа агента ---------- -def create_agent() -> StateGraph: +def ask_and_run(user_input: Optional[str], config: Dict[str, Any]) -> None: """ - Возвращает готовый граф агента с памятью и возможностью - прерывания перед вызовом инструмента. + Запускает потоковый диалог с агентом. Если агент хочет вызвать инструмент, + пользователь получает запрос на подтверждение. """ - # Состояние: хранит сообщения - def add_message(state: Dict[str, Any], message: Dict[str, str]) -> Dict[str, Any]: - state["messages"].append(message) - return state - - graph = StateGraph(add_message) - - # Входной узел – пользовательское сообщение - graph.set_entry_point("user") - - # Узлы для обработки сообщений и вызова инструментов - graph.add_node("assistant", lambda state: {"messages": [state["messages"][-1]]}) - graph.add_node("tools", tool_executor) - - # Переходы - graph.set_conditional_edges( - "assistant", - lambda _: "tools" if any(m.get("tool_calls") for m in _["messages"]) else "final", + # Если user_input – None, продолжаем с того места, где остановились + stream = agent.stream( + {"messages": [] if user_input is None else [{"role": "user", "content": user_input}]}, + config=config, + stream_mode=["messages", "updates"], ) - graph.add_edge("tools", "assistant") - graph.add_edge("assistant", "final") - # Конфигурация прерывания - graph.set_interrupt_before(["tools"]) - - return graph - - -agent = create_agent() -memory = MemorySaver() -agent.compile(checkpointer=memory) - - -# ---------- 3. Функция взаимодействия ---------- -def ask_and_run(user_input: Dict[str, Any], config: Dict[str, Any]) -> None: - """ - Запускает потоковый ответ агента с возможностью подтверждения - вызова инструмента пользователем. - """ - # Инициализируем сообщение пользователя в состоянии - if user_input is not None: - agent.send(user_input, config=config) - - for chunk_type, chunk_data in agent.stream( - {}, config=config, stream_mode=["messages", "updates"] - ): - state = agent.get_state(config) - - # 3.1 Вывод токенов LLM + for chunk_type, chunk_data in stream: + # 1. Печать токенов (части сообщения) if chunk_type == "messages": - token = chunk_data["content"] or "" - console.print(token, end="", style="cyan") + console.print(chunk_data["content"], end="", style="cyan") continue - # 3.2 Обработка обновлений (вызовы инструментов) - if chunk_type == "updates": - for update in chunk_data: - console.print(f"\n[bold magenta]Обновление:{update}") + # 2. Обработка вызова инструмента + if chunk_type == "updates" and "tool_calls" in chunk_data: + for tool_call in chunk_data["tool_calls"]: + name = tool_call["name"] + args = tool_call.get("args", {}) + console.print(f"\n[bold magenta]Tool call:[/bold magenta] {name}({args})") continue - # 3.3 Пауза перед вызовом инструмента - if "__interrupt__" in chunk_data and state.next == ("tools",): - tool_call = state.values["messages"][-1]["tool_calls"][0] + # 3. Обнаружение паузы перед инструментом + if "__interrupt__" in chunk_data and agent.get_state(config).next == ("tools",): + # Получаем информацию о том, какой инструмент будет вызван + state = agent.get_state(config) + last_msg = state.values["messages"][-1] + tool_call = last_msg.tool_calls[0] name = tool_call["name"] args = tool_call.get("args", {}) - console.print(f"\n[bold yellow]Пауза: [{name}]({args})") - answer = input("Разрешить? (Y/n): ").strip().lower() + console.print(f"\n[bold yellow]Пауза:[/bold yellow] {name}({args})") + answer = input("\nРазрешить вызов инструмента? (Y/n): ").strip().lower() if answer in ("y", ""): - # Возобновляем с того места, где остановились + # Продолжаем с того же состояния ask_and_run(None, config) else: - console.print("[red]Отменено пользователем.") - break - - -# ---------- 4. Основной цикл ---------- -def main() -> None: - thread_id = "thread-1" - config = {"configurable": {"thread_id": thread_id}} - - console.print("[bold green]Запуск агента. Введите 'exit' для выхода.\n") - - while True: - user_input = input("\nВы: ") - if user_input.lower() == "exit": + console.print("[red]Вызов отменён.[/red]") break - ask_and_run( - {"messages": [{"role": "human", "content": user_input}]}, - config, - ) - console.print("[bold green]До свидания!") +# ------------------------------------------------------------ +# Точка входа +# ------------------------------------------------------------ if __name__ == "__main__": - main() \ No newline at end of file + thread_id = "thread-1" + config = {"configurable": {"thread_id": thread_id}} + console.print("[bold green]Запуск агента. Введите сообщение, а затем Enter.[/bold green]") + while True: + try: + user_msg = input("\nВы: ") + if user_msg.lower() in ("exit", "quit"): + console.print("[bold red]Завершение работы.[/bold red]") + break + ask_and_run(user_msg, config) + except KeyboardInterrupt: + console.print("\n[red]Прервано пользователем. Выход.[/red]") + break \ No newline at end of file