Практическое задание №3: Память и подтверждение действий: solution.py
This commit is contained in:
+88
-81
@@ -1,116 +1,123 @@
|
||||
# solution.py
|
||||
|
||||
import os
|
||||
from typing import Any, Dict, Iterable, Tuple, Optional
|
||||
"""
|
||||
Основной скрипт агента с памятью и подтверждением действий.
|
||||
"""
|
||||
|
||||
from langchain.chat_models import ChatOpenAI
|
||||
from langchain.tools import BaseTool
|
||||
import os
|
||||
from typing import Any, Dict, Iterable, Tuple
|
||||
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langgraph.checkpoint.memory import MemorySaver
|
||||
from langgraph.graph import StateGraph
|
||||
from langgraph.prebuilt import create_agent
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
from rich.console import Console
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Конфигурация LLM и инструмента
|
||||
# ------------------------------------------------------------
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 1. Конфигурация 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()
|
||||
# Задаём ключ API OpenAI (можно задать переменной окружения)
|
||||
os.environ.setdefault("OPENAI_API_KEY", "sk-...")
|
||||
|
||||
# Пример простого инструмента
|
||||
class EchoTool(BaseTool):
|
||||
name: str = "echo"
|
||||
description: str = "Возвращает переданный текст без изменений."
|
||||
def _run(self, text: str) -> str:
|
||||
return f"Echo: {text}"
|
||||
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
|
||||
|
||||
my_tool = EchoTool()
|
||||
# Пример простого инструмента: поиск в интернете
|
||||
def search(query: str) -> str:
|
||||
"""Простейший инструмент поиска."""
|
||||
return f"Результаты поиска по запросу '{query}': ..."
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Создание агента с памятью и паузой
|
||||
# ------------------------------------------------------------
|
||||
search_tool = {
|
||||
"name": "search",
|
||||
"description": "Используется для поиска информации в интернете.",
|
||||
"func": search,
|
||||
}
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 2. Создаём агент с памятью и возможностью паузы перед вызовом инструмента
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
memory = MemorySaver()
|
||||
agent = create_agent(
|
||||
|
||||
agent = create_react_agent(
|
||||
model=llm,
|
||||
tools=[my_tool],
|
||||
system_prompt="Ты полезный ассистент. Перед каждым вызовом инструмента спрашивай разрешение.",
|
||||
checkpointer=memory, # нужна для сохранения состояния паузы
|
||||
interrupt_before=["tools"], # остановка перед инструментом
|
||||
tools=[search_tool],
|
||||
prompt="You are a helpful assistant.",
|
||||
checkpointer=memory, # подключаем память
|
||||
interrupt_before=["tools"], # ставим паузу перед каждым вызовом инструмента
|
||||
)
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 3. Вспомогательные функции
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
console = Console()
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Функция взаимодействия с пользователем
|
||||
# ------------------------------------------------------------
|
||||
|
||||
def ask_and_run(user_input: Optional[str], config: Dict[str, Any]) -> None:
|
||||
def ask_and_run(user_input: Dict[str, Any], config: Dict[str, Any]) -> None:
|
||||
"""
|
||||
Запускает потоковый диалог с агентом. Если агент хочет вызвать инструмент,
|
||||
пользователь получает запрос на подтверждение.
|
||||
Обрабатывает поток от агента, выводит токены и обрабатывает паузы.
|
||||
"""
|
||||
# Если user_input – None, продолжаем с того места, где остановились
|
||||
stream = agent.stream(
|
||||
{"messages": [] if user_input is None else [{"role": "user", "content": user_input}]},
|
||||
config=config,
|
||||
stream_mode=["messages", "updates"],
|
||||
)
|
||||
# Переменная для хранения последнего состояния (для возобновления)
|
||||
last_state = None
|
||||
|
||||
for chunk in agent.stream(user_input, config=config, stream_mode=["messages", "updates"]):
|
||||
chunk_type, chunk_data = chunk
|
||||
|
||||
for chunk_type, chunk_data in stream:
|
||||
# 1. Печать токенов (части сообщения)
|
||||
if chunk_type == "messages":
|
||||
console.print(chunk_data["content"], end="", style="cyan")
|
||||
continue
|
||||
# Печатаем токены ответа агента
|
||||
console.print(chunk_data["content"], end="")
|
||||
|
||||
# 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
|
||||
elif chunk_type == "updates":
|
||||
# Выводим информацию о вызове инструмента (если есть)
|
||||
for update in chunk_data:
|
||||
if isinstance(update, dict) and "tool_calls" in update.get("values", {}):
|
||||
tool_call = update["values"]["messages"][-1].tool_calls[0]
|
||||
console.print(f"\n[bold cyan]Инструмент:[/bold cyan] {tool_call['name']}({tool_call['args']})")
|
||||
|
||||
# 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]Пауза:[/bold yellow] {name}({args})")
|
||||
answer = input("\nРазрешить вызов инструмента? (Y/n): ").strip().lower()
|
||||
if answer in ("y", ""):
|
||||
# Продолжаем с того же состояния
|
||||
tool_call = state.values["messages"][-1].tool_calls[0]
|
||||
console.print(f"\n[bold magenta]Пауза:[/bold magenta] Агент хочет вызвать инструмент "
|
||||
f"[green]{tool_call['name']}[/green] с аргументами {tool_call['args']}")
|
||||
# Спрашиваем подтверждение у пользователя
|
||||
answer = input("Разрешить? (Y/n): ").strip().lower()
|
||||
if answer in ("", "y", "yes"):
|
||||
console.print("[bold green]Продолжаем...[/bold green]")
|
||||
# Возобновляем работу агента с того места, где остановились
|
||||
ask_and_run(None, config)
|
||||
return
|
||||
else:
|
||||
console.print("[red]Вызов отменён.[/red]")
|
||||
break
|
||||
console.print("[bold red]Отменено. Завершаем разговор.[/bold red]")
|
||||
break
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Точка входа
|
||||
# ------------------------------------------------------------
|
||||
last_state = chunk_data
|
||||
|
||||
if __name__ == "__main__":
|
||||
console.print("\n[bold blue]Финальный ответ получен.[/bold blue]\n")
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 4. Основной цикл чата
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def main() -> None:
|
||||
"""
|
||||
Запускает бесконечный чат с агентом.
|
||||
"""
|
||||
thread_id = "thread-1"
|
||||
config = {"configurable": {"thread_id": thread_id}}
|
||||
console.print("[bold green]Запуск агента. Введите сообщение, а затем Enter.[/bold green]")
|
||||
|
||||
console.print("[bold underline]Запуск агента. Введите 'exit' для завершения.[/bold underline]")
|
||||
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]")
|
||||
user_input = input("\nВы: ")
|
||||
if user_input.lower() == "exit":
|
||||
console.print("[bold red]Завершение работы.[/bold red]")
|
||||
break
|
||||
|
||||
# Формируем сообщение в формате, ожидаемом агентом
|
||||
message = {"messages": [{"role": "human", "content": user_input}]}
|
||||
ask_and_run(message, config)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user