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

2026-06-24 15:27:26 +03:00
parent 3e87be504b
commit f45e780aba
3 changed files with 146 additions and 143 deletions
@@ -1,33 +1,62 @@
-# Практическое задание №3: Память и подтверждение действий
+# LangGraph Agent with Memory and Tool Confirmation

-Главная
-Мои задания
-Практическое задание №3: Память и подтверждение действий
-5Д
-EN
-Практическое задание №3: Память и подтверждение действий
-Зачёт
-Версия 2
-Дедлайн сдачи: 31.08.2026
+This project demonstrates a simple conversational agent built with **LangGraph** that:

-В работе
+- **Remembers** the conversation history across turns using an in‑memory checkpoint.
+- **Pauses** before invoking any tool, asking the user for confirmation.
+- Uses the **Rich** library for pretty console output.

-Редактирование ответа
+## Prerequisites

-Заполните ответ и отправьте работу на проверку преподавателю.
+- Python 3.10+
+- An OpenAI API key (set as the `OPENAI_API_KEY` environment variable).

-Тип ответа
-Текст
-Ссылка
-Файлы
-Ссылка (URL)
-Прикреплённые файлы
-Загрузить файл
-Отправить на проверку
-Отменить
+## Installation

-Задание
+```bash
+# Clone the repository (or copy the files)
+git clone https://github.com/yourusername/langgraph-agent.git
+cd langgraph-agent

-Цель
+# Create a virtual environment (optional but recommended)
+python -m venv .venv
+source .venv/bin/activate   # On Windows: .venv\Scripts\activate

-Доработать агента из предыдущих заданий: добавить память разговора и механизм подтверждения каждо
+# Install dependencies
+pip install -r requirements.txt
+```
+
+## Usage
+
+```bash
+python src/main.py
+```
+
+You will see a prompt:
+
+```
+Start chat. Type 'exit' to quit.
+You:
+```
+
+Type any message. If the agent decides to use a tool (e.g., the `echo` tool), it will pause and ask:
+
+```
+Agent wants to call tool: echo({"text":"Hello"})
+Разрешить? (Y/n):
+```
+
+- Type `Y` or press Enter to allow the tool to run.
+- Type `n` to cancel the tool call.
+
+The conversation history is preserved across turns, so the agent can refer back to earlier messages.
+
+## Customizing
+
+- **Tools**: Add more tools by defining functions decorated with `@tool` from `langchain.tools`.
+- **Memory**: Replace `MemorySaver()` with a persistent checkpoint (e.g., `RedisSaver`) for long‑term storage.
+- **Prompt**: Modify the `system_prompt` in `create_agent` to change the agent’s behavior.
+
+## License
+
+MIT License
@@ -1,4 +1,5 @@
-rich
-langgraph
-langchain
-openai
+langgraph==0.0.41
+langchain==0.1.12
+langchain-openai==0.0.7
+rich==13.7.1
+openai==1.12.0
@@ -1,160 +1,133 @@
-#!/usr/bin/env python3
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
 """
-A simple LangGraph agent with memory and user confirmation before tool calls.
+A simple LangGraph agent with conversation memory and tool usage confirmation.
 """

 import os
-import sys
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Iterable, Tuple

-from langgraph import create_agent
 from langgraph.checkpoint.memory import MemorySaver
-from langgraph.tools import Tool
-from langchain.chat_models import ChatOpenAI
+from langgraph.prebuilt import create_agent
+from langgraph.graph import END
+from langchain_openai import ChatOpenAI
+from langchain.tools import tool
 from rich.console import Console

-# --------------------------------------------------------------------------- #
-# Configuration
-# --------------------------------------------------------------------------- #
-
-# Ensure the OpenAI API key is set
-OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
-if not OPENAI_API_KEY:
-    print("Error: OPENAI_API_KEY environment variable not set.")
-    sys.exit(1)
-
-# Rich console for pretty output
+# Initialize Rich console
 console = Console()

-# LLM model
-llm = ChatOpenAI(temperature=0, openai_api_key=OPENAI_API_KEY)
+# Ensure OpenAI API key is set
+if "OPENAI_API_KEY" not in os.environ:
+    console.print("[red]Error:[/red] OPENAI_API_KEY environment variable not set.")
+    console.print("Please set it before running the script.")
+    exit(1)

-# Memory saver for conversation persistence
+# Define a simple echo tool
+@tool
+def echo(text: str) -> str:
+    """
+    Echo the input text back to the user.
+    """
+    return f"Echo: {text}"
+
+# Initialize the LLM
+llm = ChatOpenAI(temperature=0)
+
+# Initialize memory saver
 memory = MemorySaver()

-# --------------------------------------------------------------------------- #
-# Tool definition
-# --------------------------------------------------------------------------- #
-
-def echo_tool(message: str) -> str:
-    """
-    A simple echo tool that returns the message back to the user.
-    """
-    return f"Echo: {message}"
-
-# Wrap the function as a LangGraph Tool
-echo = Tool.from_function(
-    fn=echo_tool,
-    name="echo",
-    description="Echoes back the provided message."
-)
-
-# --------------------------------------------------------------------------- #
-# Agent creation
-# --------------------------------------------------------------------------- #
-
-system_prompt = """
-You are a helpful assistant. When you need to use a tool, you will call it.
-"""
-
+# Create the agent with interrupt_before to pause before tool calls
 agent = create_agent(
    model=llm,
    tools=[echo],
-    system_prompt=system_prompt,
-    checkpointer=memory,          # Enable memory
-    interrupt_before=["tools"],   # Pause before any tool call
+    system_prompt=(
+        "You are a helpful assistant. "
+        "When you need to use a tool, you will be paused for confirmation."
+    ),
+    checkpointer=memory,
+    interrupt_before=["tools"],
 )

-# --------------------------------------------------------------------------- #
-# Conversation loop
-# --------------------------------------------------------------------------- #
-
-def ask_and_run(user_input: Optional[str], config: Dict[str, Any]) -> None:
+def ask_and_run(user_input: str | None, config: Dict[str, Any]) -> None:
    """
-    Handles streaming from the agent, pauses before tool calls, and asks the user
-    for confirmation before executing the tool.
+    Send user input to the agent, handle tool call interruptions,
+    and recursively resume or cancel based on user confirmation.
    """
-    # If user_input is provided, send it as a new message
+    # Prepare the input payload
+    payload = {"messages": []}
    if user_input is not None:
-        # The agent expects a dict with a "messages" key
-        input_payload = {"messages": [{"role": "user", "content": user_input}]}
-    else:
-        # None means resume from the paused state
-        input_payload = {}
+        payload["messages"].append({"role": "user", "content": user_input})

    # Stream the agent's response
-    for chunk_type, chunk_data in agent.stream(
-        input_payload,
+    for chunk in agent.stream(
+        payload,
        config=config,
        stream_mode=["messages", "updates"],
    ):
+        state = agent.get_state(config)
+        chunk_type, chunk_data = chunk
+
        # Handle message chunks
        if chunk_type == "messages":
-            # chunk_data is a list of messages; print the last one
-            if isinstance(chunk_data, list) and chunk_data:
-                last_msg = chunk_data[-1]
-                if last_msg.get("role") == "assistant":
-                    console.print(f"[bold cyan]Assistant:[/bold cyan] {last_msg.get('content', '')}")
-                elif last_msg.get("role") == "tool":
-                    console.print(f"[bold magenta]Tool Output:[/bold magenta] {last_msg.get('content', '')}")
-                else:
-                    console.print(last_msg.get("content", ""))
-            else:
-                console.print(chunk_data)
+            for msg in chunk_data:
+                role = msg.get("role")
+                content = msg.get("content", "")
+                if role == "assistant":
+                    console.print(f"[bold green]Assistant:[/bold green] {content}")
+                elif role == "user":
+                    console.print(f"[bold blue]User:[/bold blue] {content}")

-        # Detect interrupt before tool call
-        if "__interrupt__" in chunk_data and agent.get_state(config).next == ("tools",):
-            # Retrieve the pending tool call
-            state = agent.get_state(config)
-            try:
-                last_message = state.values["messages"][-1]
-                tool_call = last_message.tool_calls[0]
-                tool_name = tool_call["name"]
-                tool_args = tool_call["args"]
-                console.print(f"[yellow]Agent wants to call tool:[/yellow] {tool_name}({tool_args})")
-            except Exception as e:
-                console.print(f"[red]Error retrieving tool call: {e}[/red]")
-                break
+        # Detect interruption before tool call
+        if "__interrupt__" in chunk_data and state.next == ("tools",):
+            # Extract the pending tool call
+            last_msg = state.values["messages"][-1]
+            tool_calls = last_msg.get("tool_calls", [])
+            if not tool_calls:
+                console.print("[red]Error:[/red] No tool call found during interruption.")
+                return

-            # Ask user for confirmation
-            console.print("[bold]Allow tool execution? (Y/n):[/bold] ", end="")
-            answer = input().strip().lower()
-            if answer in ("", "y", "yes"):
-                console.print("[green]Executing tool...[/green]")
-                # Recursively resume the agent
+            tool_call = tool_calls[0]
+            tool_name = tool_call.get("name")
+            tool_args = tool_call.get("args", {})
+
+            console.print(
+                f"[yellow]Agent wants to call tool:[/yellow] {tool_name}({tool_args})"
+            )
+            answer = console.input("Разрешить? (Y/n): ")
+
+            if answer.lower().strip() in ("", "y", "yes"):
+                # Resume the agent from the interruption point
                ask_and_run(None, config)
+                return
            else:
-                console.print("[red]Tool execution cancelled by user.[/red]")
-                break
+                console.print("[red]Отменено[/red]")
+                return

-        # Handle updates (optional)
+        # Handle updates (e.g., tool results) if needed
        if chunk_type == "updates":
-            # For this simple example, we ignore updates
+            # In this simple example we don't process updates separately
            pass

+        # If the agent has finished, exit the loop
+        if state.next == END:
+            return

 def main() -> None:
    """
-    Main conversation loop.
+    Main chat loop.
    """
-    # Use a fixed thread ID for this session
-    config = {"configurable": {"thread_id": "thread-1"}}
-
-    console.print("[bold green]Welcome to the LangGraph Agent![/bold green]")
-    console.print("Type your messages below. Press Ctrl+C to exit.\n")
+    thread_id = "thread-1"
+    config = {"configurable": {"thread_id": thread_id}}

+    console.print("[bold cyan]Start chat. Type 'exit' to quit.[/bold cyan]")
    while True:
-        try:
-            user_input = input("[bold]You:[/bold] ")
-            if not user_input:
-                continue
-            ask_and_run(user_input, config)
-        except KeyboardInterrupt:
-            console.print("\n[bold red]Exiting...[/bold red]")
+        user_input = console.input("[bold blue]You:[/bold blue] ")
+        if user_input.lower() in ("exit", "quit"):
+            console.print("[bold magenta]Goodbye![/bold magenta]")
            break
-        except Exception as e:
-            console.print(f"[red]Unexpected error: {e}[/red]")
-
+        ask_and_run(user_input, config)

 if __name__ == "__main__":
    main()