Tworzenie Systemu opartego na Agentach AI z LangGraph: Integracja Człowieka w Proces Decyzyjny
Tworzenie Systemu Agentów AI z Możliwością Interakcji z Człowiekiem
W poprzednim poradniku omówiliśmy budowę agenta AI, który potrafił odpowiadać na zapytania, przeszukując internet oraz przechowywać swój stan między kolejnymi interakcjami. Jednak w wielu przypadkach konieczne jest wprowadzenie człowieka do procesu decyzyjnego, aby monitorować i zatwierdzać działania agenta. Dzięki LangGraph można to łatwo osiągnąć. W tym artykule omówimy, jak dodać człowieka do pętli decyzyjnej i umożliwić mu kontrolowanie działań agenta.
—
Konfiguracja Środowiska
Zaczniemy od przygotowania środowiska, definiując zmienne środowiskowe, importując niezbędne biblioteki oraz konfigurując mechanizm zapisu stanu agenta.
Instalacja wymaganych pakietów
Najpierw należy zainstalować niezbędne biblioteki:
bash
pip install langgraph==0.2.53 langgraph-checkpoint==2.0.6 langgraph-sdk==0.1.36 langchain-groq langchain-community langgraph-checkpoint-sqlite==2.0.1
Konfiguracja kluczy API
Aby działać z niektórymi usługami, potrzebujemy kluczy API:
python
import os
os.environ['TAVILY_API_KEY'] = ""
os.environ['GROQ_API_KEY'] = ""
Ustawienie mechanizmu zapisu stanu
Zapisywanie stanu agenta w bazie SQLite pozwala na trwałość jego działań i przerywanie oraz wznawianie procesów:
python
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
sqlite_conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
memory = SqliteSaver(sqlite_conn)
Definiowanie Agenta
Agent AI będzie operował na podstawie modelu językowego i wykorzystywał narzędzia do wyszukiwania informacji.
python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage, AIMessage
from langchain_groq import ChatGroq
from langchain_community.tools.tavily_search import TavilySearchResults
Następnie definiujemy klasę agenta:
python
class Agent:
def __init__(self, model, tools, checkpointer, system=""):
self.system = system
graph = StateGraph(AgentState)
graph.add_node("llm", self.call_openai)
graph.add_node("action", self.take_action)
graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
graph.add_edge("action", "llm")
graph.set_entry_point("llm")
self.graph = graph.compile(checkpointer=checkpointer)
self.tools = {t.name: t for t in tools}
self.model = model.bind_tools(tools)
def call_openai(self, state: AgentState):
messages = state['messages']
if self.system:
messages = [SystemMessage(content=self.system)] + messages
message = self.model.invoke(messages)
return {'messages': [message]}
def exists_action(self, state: AgentState):
result = state['messages'][-1]
return len(result.tool_calls) > 0
def take_action(self, state: AgentState):
tool_calls = state['messages'][-1].tool_calls
results = []
for t in tool_calls:
result = self.tools[t['name']].invoke(t['args'])
results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
return {'messages': results}
—
Konfiguracja Stanu Agenta
Agent przechowuje historię rozmów i wyników, ale w tym przypadku chcemy umożliwić użytkownikowi edytowanie wiadomości zamiast jedynie ich dodawania.
python
from uuid import uuid4
def reduce_messages(left: list[AnyMessage], right: list[AnyMessage]) -> list[AnyMessage]:
for message in right:
if not message.id:
message.id = str(uuid4())
merged = left.copy()
for message in right:
for i, existing in enumerate(merged):
if existing.id == message.id:
merged[i] = message
break
else:
merged.append(message)
return merged
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], reduce_messages]
—
Dodanie Człowieka do Procesu Decyzyjnego
Aby umożliwić człowiekowi zatwierdzanie decyzji agenta, dodajemy mechanizm przerywania przed wykonaniem konkretnej akcji.
python
class Agent:
def __init__(self, model, tools, checkpointer, system=""):
self.graph = graph.compile(checkpointer=checkpointer, interrupt_before=["action"])
—
Uruchamianie Agenta
Teraz możemy uruchomić agenta, wykorzystując model językowy i narzędzie do wyszukiwania informacji.
python
prompt = """Jesteś inteligentnym asystentem badawczym. Korzystaj z wyszukiwarki internetowej, aby znaleźć informacje.
Możesz wykonywać wiele zapytań w razie potrzeby. Wyszukuj informacje tylko wtedy, gdy masz pewność, czego szukasz."""
model = ChatGroq(model="Llama-3.3-70b-Specdec")
tool = TavilySearchResults(max_results=2)
abot = Agent(model, [tool], system=prompt, checkpointer=memory)
messages = [HumanMessage(content="Jaka jest pogoda w Warszawie?")]
thread = {"configurable": {"thread_id": "1"}}
for event in abot.graph.stream({"messages": messages}, thread):
for v in event.values():
print(v)
Podczas działania agenta użytkownik może przerywać proces i zatwierdzać kolejne kroki ręcznie.
—
Interaktywne Zatwierdzanie przez Człowieka
Wprowadzamy pętlę, która pozwala użytkownikowi ręcznie akceptować lub odrzucać decyzje agenta:
python
messages = [HumanMessage("Jaka jest pogoda w Krakowie?")]
thread = {"configurable": {"thread_id": "2"}}
for event in abot.graph.stream({"messages": messages}, thread):
for v in event.values():
print(v)
while abot.graph.get_state(thread).next:
print("n", abot.graph.get_state(thread), "n")
_input = input("Kontynuować? (t/n): ")
if _input.lower() != "t":
print("Przerwano")
break
for event in abot.graph.stream(None, thread):
for v in event.values():
print(v)
—
Podsumowanie
Dzięki LangGraph możemy nie tylko stworzyć inteligentnego agenta AI, ale także umożliwić człowiekowi nadzorowanie jego działań. Taki system może być wykorzystywany w różnych zastosowaniach, od obsługi klienta po analizy danych i badania naukowe. Spróbuj eksperymentować z różnymi punktami przerwania i sprawdź, jak zmienia się zachowanie agenta w zależności od ingerencji użytkownika.
