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

2026-06-24 15:49:40 +03:00
parent af6701671b
commit ddd2bdb8a9
3 changed files with 99 additions and 131 deletions
@@ -1,33 +1,48 @@
-# Практическое задание №3: Память и подтверждение действий
+# Interactive LangGraph Agent with Memory and Tool Confirmation

-Главная
-Мои задания
-Практическое задание №3: Память и подтверждение действий
-5Д
-EN
-Практическое задание №3: Память и подтверждение действий
-Зачёт
-Версия 4
-Дедлайн сдачи: 31.08.2026
+This project demonstrates a simple interactive agent built with LangGraph that:
+- Maintains conversation memory across turns.
+- Pauses before invoking tools and asks the user for confirmation.
+- Uses the `rich` library for pretty console output.

-В работе
+## Prerequisites

-Редактирование ответа
+- Python 3.10 or newer
+- An OpenAI API key (set as the environment variable `OPENAI_API_KEY`)

-Заполните ответ и отправьте работу на проверку преподавателю.
+## Installation

-Тип ответа
-Текст
-Ссылка
-Файлы
-Ссылка (URL)
-Прикреплённые файлы
-Загрузить файл
-Отправить на проверку
-Отменить
+```bash
+# Clone the repository
+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
+# Ensure your OpenAI API key is set
+export OPENAI_API_KEY="sk-..."
+
+# Run the agent
+python src/main.py
+```
+
+You will see a prompt where you can type messages. The agent will respond, and if it needs to use the calculator tool, it will pause and ask for your confirmation before executing the tool.
+
+Type `exit` to quit the program.
+
+## How It Works
+
+- **Memory**: `MemorySaver` stores the conversation history, allowing the agent to remember previous messages.
+- **Interrupt Before Tool Calls**: The agent is configured with `interrupt_before=['tools']`, causing it to pause before calling any tool.
+- **User Confirmation**: When the agent pauses, it prints the pending tool call and asks you whether to proceed. If you confirm, the agent resumes; otherwise, the tool call is cancelled.
+
+Feel free to extend the toolset or modify the system prompt to suit your needs.
@@ -1,5 +1,4 @@
-langgraph==0.0.41
-langchain==0.1.12
-langchain-openai==0.0.7
-rich==13.7.1
-openai==1.12.0
+langgraph
+langchain-openai
+rich
+openai
@@ -1,131 +1,85 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-"""
-A simple LangGraph agent with conversation memory and tool usage confirmation.
-"""
-
-import os
-from typing import Any, Dict, Iterable, Tuple
-
+from langgraph import create_agent
+from langgraph.tools import tool
 from langgraph.checkpoint.memory import MemorySaver
-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

-# Initialize Rich console
 console = Console()

-# 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)
+# Define a simple calculator tool
+@tool(name="calculator", description="Evaluates an arithmetic expression.")
+def calculator(expression: str) -> str:
+    try:
+        # Use eval in a safe context
+        result = eval(expression, {"__builtins__": None}, {})
+        return str(result)
+    except Exception as e:
+        return f"Error evaluating expression: {e}"

-# Define a simple echo tool
-@tool
-def echo(text: str) -> str:
-    """
-    Echo the input text back to the user.
-    """
-    return f"Echo: {text}"
+# Initialize LLM
+llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

-# Initialize the LLM
-llm = ChatOpenAI(temperature=0)
-
-# Initialize memory saver
+# Memory for conversation
 memory = MemorySaver()

-# Create the agent with interrupt_before to pause before tool calls
+# Create agent with interrupt_before to pause before tool calls
 agent = create_agent(
-    model=llm,
-    tools=[echo],
-    system_prompt=(
-        "You are a helpful assistant. "
-        "When you need to use a tool, you will be paused for confirmation."
-    ),
+    llm=llm,
+    tools=[calculator],
+    system_prompt="You are a helpful assistant. Use the calculator tool to evaluate expressions.",
    checkpointer=memory,
    interrupt_before=["tools"],
 )

-def ask_and_run(user_input: str | None, config: Dict[str, Any]) -> None:
+def ask_and_run(user_input: str | None, config: dict):
    """
-    Send user input to the agent, handle tool call interruptions,
-    and recursively resume or cancel based on user confirmation.
+    Run the agent with optional user input. Handles pauses before tool calls
+    and asks the user for confirmation before executing a tool.
    """
-    # Prepare the input payload
-    payload = {"messages": []}
+    # Prepare input for the agent
    if user_input is not None:
-        payload["messages"].append({"role": "user", "content": user_input})
+        input_dict = {"messages": [{"role": "user", "content": user_input}]}
+    else:
+        input_dict = {}

-    # Stream the agent's response
-    for chunk in agent.stream(
-        payload,
-        config=config,
-        stream_mode=["messages", "updates"],
+    for chunk_type, chunk_data in agent.stream(
+        input_dict, config=config, stream_mode=["messages", "updates"]
    ):
-        state = agent.get_state(config)
-        chunk_type, chunk_data = chunk
-
        # Handle message chunks
        if chunk_type == "messages":
-            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 interruption before tool call
-        if "__interrupt__" in chunk_data and state.next == ("tools",):
-            # Extract the pending tool call
+            # Check for 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)
                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
-
-            tool_call = tool_calls[0]
-            tool_name = tool_call.get("name")
-            tool_args = tool_call.get("args", {})
-
+                tool_call = last_msg.tool_calls[0]
                console.print(
-                f"[yellow]Agent wants to call tool:[/yellow] {tool_name}({tool_args})"
+                    f"[bold yellow]Tool call requested:[/bold yellow] {tool_call['name']}({tool_call['args']})"
                )
-            answer = console.input("Разрешить? (Y/n): ")
-
-            if answer.lower().strip() in ("", "y", "yes"):
-                # Resume the agent from the interruption point
+                answer = input("Разрешить? (Y/n): ")
+                if answer.lower().strip() == "y":
+                    # Resume the agent from the same state
                    ask_and_run(None, config)
-                return
                else:
-                console.print("[red]Отменено[/red]")
-                return
+                    console.print("[red]Tool call cancelled.[/red]")
+                break  # exit the current stream loop

-        # Handle updates (e.g., tool results) if needed
-        if chunk_type == "updates":
-            # In this simple example we don't process updates separately
+            # Print the messages
+            for msg in chunk_data.get("messages", []):
+                console.print(f"[{msg['role']}] {msg['content']}")
+
+        # Handle updates (optional)
+        elif chunk_type == "updates":
+            # For simplicity, we ignore updates in this example
            pass

-        # If the agent has finished, exit the loop
-        if state.next == END:
-            return
-
-def main() -> None:
-    """
-    Main chat loop.
-    """
-    thread_id = "thread-1"
-    config = {"configurable": {"thread_id": thread_id}}
-
-    console.print("[bold cyan]Start chat. Type 'exit' to quit.[/bold cyan]")
+def main():
+    config = {"configurable": {"thread_id": "conversation-1"}}
+    console.print("[bold green]Welcome to the interactive agent. Type 'exit' to quit.[/bold green]")
    while True:
-        user_input = console.input("[bold blue]You:[/bold blue] ")
-        if user_input.lower() in ("exit", "quit"):
-            console.print("[bold magenta]Goodbye![/bold magenta]")
+        user_input = input("> ")
+        if user_input.lower() == "exit":
+            console.print("[bold blue]Goodbye![/bold blue]")
            break
        ask_and_run(user_input, config)