Zanim zaczniemy, celem lepszego zrozumienia występujących w artykule pojęć, zachęcam zapoznanie się z kilkoma kluczowymi definicjami:
- LLM (Large Language Model) – duży model językowy, czyli algorytm głębokiego uczenia, który może wykonywać zadania przetwarzania języka naturalnego (NLP).
- Prompt – zapytanie w języku naturalnym przekazywane do LLM-a.
- Augmented LLM (Augmented Large Language Model) – rozszerzony model językowy, który poza standardowym operowaniem na własnej bazie wiedzy może używać dodatkowych narzędzi, np. bazy wiedzy firmy, w celu poprawy jakości odpowiedzi.
- RAG (Retrieval-Augmented Generation) – technika, w której model językowy generuje odpowiedzi na podstawie dokumentów pozyskanych z zewnętrznych baz wiedzy.
- Agent AI – autonomiczny system AI podejmujący decyzje i wykonujący akcje z wykorzystaniem modeli językowych i narzędzi.
- Graf – oznacza proces (tzw. workflow) w LangGraph, składający się z węzłów oraz ich powiązań (akcji).
- Orkiestracja – zaawansowany proces zarządzania, koordynowania i kontrolowania procesów w sposób, który zapewnia ich spójne i harmonijne działanie.
Wprowadzenie do Human-in-the-loop
Koncepcja Human-in-the-loop (w skrócie HILP) oznacza zaangażowanie człowieka w procesy decyzyjne systemów AI.
Pomimo że HILP funkcjonuje w świecie technologii od dawna, to właśnie tu – tj. w dziedzinie AI – ten proces staje się wyjątkowo użyteczny. Dzieje się tak dlatego, że systemy oparte na sztucznej inteligencji – pomimo że obecnie są już niezwykle skuteczne – wciąż mają scenariusze, w których ludzka ocena i weryfikacja są wręcz niezbędne. W szczególności, gdy chodzi o precyzję i bezpieczeństwo wyników.
Jak działa HILP w praktyce?
System AI, w momencie napotkania scenariusza, który nie został zawarty w zestawie reguł, zatrzymuje proces (a dokładniej – pętlę procesu), oczekując na odpowiedź od człowieka. Dzięki zatwierdzeniu lub wprowadzeniu poprawek – Agent kontynuuje pracę w określonym przez „ludzkiego nadzorcę” kierunku.
LangGraph, czyli budowanie AI-owych procesów
Aplikacje Agentów AI można budować z użyciem podstawowych bibliotek (SDK) udostępnianych przez firmy tworzące modele językowe. Obecnie jedną z popularniejszych jest SDK firmy OpenAI – odpowiedzialną m.in. za model ChatGPT.
Idąc krok dalej, wraz z rozwojem oprogramowania opartego o LLM-y powstały dodatkowe frameworki, które znacznie rozszerzyły możliwości bibliotek standardowych.
Frameworki pozwalają na łączenie się z modelami różnych producentów oraz zapewniają szereg dodatkowych funkcji ułatwiających budowanie aplikacji AI, jak również ich orkiestrację.
Jednym z popularniejszych na chwilę obecną jest LangGraph – framework umożliwiający budowanie zaawansowanych procesów z użyciem LLM-ów. LangGraph pozwala w relatywnie prosty sposób włączać różne komponenty: narzędzia, pamięć zewnętrzną. Ułatwia też implementację koncepcji takich jak HILP – wszystko to w ramach konkretnego procesu (tzw. grafu).
Czym jest graf procesu?
LangGraph pozwala definiować etapy procesu w formie grafu, który składa się z węzłów o różnym przeznaczeniu.
Oto kilka z nich, których użyjemy w naszej aplikacji:
- Węzeł agenta – odpowiedzialny za komunikację z LLM-em (przekazywanie danych z/do LLM-a).
- Węzeł narzędzi – przykładowo narzędzie przeszukujące dodatkową bazę wiedzy.
- Węzeł decyzji człowieka – gdzie użytkownik wpływa na proces, kontrolując ścieżkę, którą aplikacja ma podążać.
Technika RAG
Istnieje jeszcze jedna ważna technika, niezbędna, aby stworzyć nasz Augmented LLM – jest nią technika RAG.
RAG rozszerza możliwości wbudowanej/treningowej bazy wiedzy LLM-a poprzez dodanie dodatkowego wyszukiwania, które pozwala modelowi na dostęp do zewnętrznych źródeł danych. Dzięki tej technice model może generować dokładniejsze odpowiedzi, ponieważ uwzględnia kontekst dodatkowych danych, niedostępnych nigdzie indziej – przykładowo danymi może być wewnętrzny FAQ organizacji.
Ostatecznie pozyskanymi danymi możemy zasilać aplikację ChatBota AI, aby udzielić precyzyjnych odpowiedzi pracownikom firmy.
W naszym kodzie RAG umożliwia wyszukanie dokumentów zaindeksowanych w bazie wektorowej.
Sama tematyka baz wektorowych jest bardzo szeroki i znacznie wykracza poza zakres obecnego artykułu. Jednakże, warto zapamiętać, że bazy wektorowe są niezbędne, abyśmy mogli skonwertować tekst na wektory, czyli matematyczne reprezentacje danych, aby ostatecznie przekazać je przez aplikację AI do LLM-a.
Implementacja Human-in-the-loop w LangGraph
Obecnie LangGraph jest frameworkiem udostępniającym SDK dla dwóch języków programowania – Python oraz JavaScript. W naszej aplikacji użyliśmy języka Python.
Dodatkowo „mózg” naszego systemu stanowi model językowy ChatGPT w wersji 4o-mini – stąd, aby uruchomić aplikację, należy wygenerować klucz API, na dedykowanej platformie firmy OpenAI (w tym celu będzie wymagane założenie konta użytkownika, jeżeli go nie posiadasz).
Co więcej, użycie API OpenAI może wiązać się również z niewielkimi kosztami, jeżeli NIE zdecydujemy się udostępniać danych naszej komunikacji z modelem, dla celów treningowych. Więcej informacji znajdziesz tutaj.
Konfiguracja środowiska developerskiego
- zainstalowany interpreter Python – w wersji 3.1.X lub nowszej
- zainstalowany manager pakietów PIP
- w tym pakiety:
- pip install chromadb + pip install langchain-unstructured
- w tym pakiety:
- klucz API z platformy OpenAI > https://platform.openai.com/api-keys
- klucz umieszczamy w pliku .env (przykład pliku znajdziesz na końcu artykułu)
- kolejny plik to plik tekstowy z rozszerzoną bazą wiedzy knowledge-base.txt – również załączony na dole artykułu
- plik umieszczamy w katalogu, w którym znajduje się aplikacja
- (opcjonalnie), jeżeli mamy potrzebę użycia Debugera, należy zainstalować bibliotekę python3.10-dev
Poniżej zalecana organizacja plików w aplikacji:
|-- /projects/hitl-example-app/
|
|-- .env (zmienne środowiskowe)
|-- app-hitl.py (kod aplikacji)
|-- knowledge-base.txt (lokalna baza wiedzy)
Omówienie kodu programu
W pierwszej kolejności importujemy niezbędne biblioteki i pakiety (sekcje import i from).
"""
Base Python libraries
"""
import os
import uuid
from pathlib import Path
from typing import Literal
from dotenv import load_dotenv
from IPython.display import Image, display
"""
Required LangGraph and LangChain libraries
"""
from langchain_core.messages import SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.types import Command, interrupt
"""
Context search libraries
"""
from langchain_community.vectorstores import Chroma
from langchain_community.vectorstores.utils import filter_complex_metadata
from langchain_unstructured import UnstructuredLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
Następnie przeprowadzamy podstawową konfigurację skryptu:
"""
-------------
Configuration
-------------
Ustawiamy tryb rozbudowanego logowania komunikatów (debug)
"""
detailed_model_response = False
"""
Ładujemy niezbędne zmienne środowiskowe jak klucz API itp.
"""
load_dotenv(dotenv_path=Path(".env"))
"""
--------------------------------------------------------------------------------------------------------------------
Data source / Knowledge Base configuration
--------------------------------------------------------------------------------------------------------------------
Wskazujemy nazwę/lokalizację pliku bazy wiedzy, która ma być zamieniona na wektory i udostępniona dla LLM-a
"""
datasource = 'knowledge-base.txt'
"""
--------------------------------------------------------------------------------------------------------------------
Build memory
--------------------------------------------------------------------------------------------------------------------
Definiujemy checkpointer, który przechowuje stanu grafu (wiadomości, informacje o aktualnym węźle itp.)
WAŻNE: W przykładzie zastosowaliśmy nietrwały zapis stanu do pamięci operacyjnej.
"""
memory = MemorySaver()
Kolejny krok stanowi konfiguracja modelu językowego i narzędzi, z których będzie korzystać LLM:
"""
Definiujemy funkcję narzędzia (anotacja `@tool`) do przeszukania bazy wiedzy
Funkcja przyjmuje zapytanie bezposrednio z LLM-a (z użyciem parametru `query`).
Każdorazowe jej wywołanie wiąże się z załadowaniem danych z pliku, ich podziałem na części i ostateczną konwersją do
postaci wektorowej.
"""
@tool
def context_searcher(query: str):
"""Search the relevant documents"""
loader = UnstructuredLoader(datasource)
document = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=50)
split_documents = text_splitter.split_documents(document)
filtered_documents = filter_complex_metadata(split_documents)
vectorstore = Chroma.from_documents(
documents=filtered_documents,
collection_name="knowledge-base",
embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
results = retriever.invoke(query)
return "\n".join([doc.page_content for doc in results])
"""
--------------------------------------------------------------------------------------------------------------------
AI Agent
--------------------------------------------------------------------------------------------------------------------
Definiujemy listę dodatkowych narzędzi, z których bedzie mogl skorzystać LLM.
Obecnie do dyspozycji mamy jedno narzędzie o nazwie `context_searcher`, które rozszerza wbudowaną bazę wiedzy LLM-a.
W ten sposób implementujemy wspominaną wcześniej technikę RAG.
"""
tools = [context_searcher]
"""
Inicjujemy LLM-a i dołączamy dostępne narzędzia.
"""
openai_api_key = os.getenv("OPENAI_API_KEY")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=openai_api_key).bind_tools(tools)
"""
WAŻNE: kluczowy w budowaniu agentów jest parametr `temperature`, który kontroluje poziom losowości generowanych
odpowiedzi.
Przy aplikacjach opierających się na własnej bazie wiedzy zaleca się podejście zero-jedynkowe, rozumiane tutaj jako
najniższy udział losowości - w przypadku ChatGPT reprezentowany jest on przez wartość temperature=0.
"""
Czas na zdefiniowanie funkcji używanych w grafie/workflow, które wprowadzą koncepcje HILP w życie:
"""
human_interaction() - to najważniejsza funkcja, w kontekście HILP ponieważ zatrzymuje workflow procesu, w oczekiwaniu
na interakcje z człowiekiem.
To co jest istotne to fakt, ze każda funkcja używana w grafie ma zdefiniowane ścieżki przejścia do innych etapów
procesu.
Definicję możliwych kroków stanowi tu `Command[Literal["human_approved", "human_rejected", END]]`, gdzie w zależności
od decyzji człowieka, możemy przejść do, któregokolwiek z węzłów ("human_approved", "human_rejected") lub zakończyć
proces (END).
"""
def human_interaction(state: MessagesState) -> Command[Literal["human_approved", "human_rejected", END]]:
"""
WAŻNE: Udostępniona przez LangGraph funkcja `interrupt()` pozwala, zatrzymać proces w celu zadania pytania
operatorowi aplikacji.
"""
answer = interrupt(
{
"question": "Hi human! :) If the answer are correct? Type: `y` or `n`",
}
)
print(">>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n")
print("Your answer: ", answer, "\n\n")
"""
W zależności od otrzymanej odpowiedzi przekierowujemy proces za pomocą akcji Command do konkretnego węzła w grafie
"""
if answer == "y":
return Command(goto="human_approved")
if answer == "n":
return Command(goto="human_rejected")
else:
print("Unsupported answer. Terminating...")
return Command(goto=END)
def human_approved(state: MessagesState) -> Command[END]:
print(">>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n")
print("✅ Do something in approved path.")
return Command(goto=END)
def human_rejected(state: MessagesState) -> Command[END]:
print(">>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n")
print("❌ Do something in rejected path.")
return Command(goto=END)
Na tym etapie mamy już funkcje dla węzłów, niezbędnych do zrealizowania koncepcji HILP i tym samym możemy przejść do modelowania kompletnego grafu:
"""
--------------------------------------------------------------------------------------------------------------------
Build state graph workflow
--------------------------------------------------------------------------------------------------------------------
WAŻNE: W tym miejscu definiujemy kilka funkcji dodatkowych używanych tylko przez Agenta AI
call_model() - inicjuje przesłanie zapytania (prompt-u) do modelu oraz zapisuje odpowiedzi z modelu do pamięci
aplikacji (pamięci zdefiniowanej, na etapie konfiguracji - w zmiennej `memory`)
"""
def call_model(state: MessagesState):
messages = state['messages']
response = model.invoke(messages)
return {"messages": [response]}
"""
Poniższa funkcja to z kolei funkcja sterująca (tzw. węzeł Conditional edge w LangGraph), która po otrzymaniu
informacji zwrotnej z LLM-a decyduje o potrzebie użycia narzędzia lub interakcji z człowiekiem
"""
def should_continue(state: MessagesState) -> Literal["tools", "human_interaction"]:
messages = state['messages']
last_message = messages[-1]
"""
Jeżeli LLM zadecyduje o potrzebie użycia narzędzi wówczas w aplikacji zwracamy nazwę węzła z dostępnymi narzędziami
"""
if last_message.tool_calls:
return "tools"
return "human_interaction"
"""
Inicjujemy graf (workflow)
"""
graph_builder = StateGraph(MessagesState)
"""
Dodajemy węzeł Agenta
"""
graph_builder.add_node("agent", call_model)
"""
Dodajemy węzeł narzędzi, które mogą być użyte przez LLM-a
"""
graph_builder.add_node("tools", ToolNode(tools))
"""
Dodajemy węzły niezbędne do realizacji koncepcji HILP
"""
graph_builder.add_node("human_interaction", human_interaction)
graph_builder.add_node("human_approved", human_approved)
graph_builder.add_node("human_rejected", human_rejected)
"""
Ostatecznie konfigurujemy punkt wejścia/wyjścia (START, END) procesu oraz węzeł sterujący (tzw. conditional edge),
pozwalający na decyzję o kolejnym etapie.
"""
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges("agent", should_continue)
graph_builder.add_edge("tools", "agent")
Zaprogramowany w ten sposób graf będzie wyglądał tak jak poniżej:
Na tym etapie nie pozostaje nam już nic innego, jak skompilowanie grafu i uruchomienie aplikacji:
"""
--------------------------------------------------------------------------------------------------------------------
Compile and run
--------------------------------------------------------------------------------------------------------------------
Kompilujemy skonfigurowany wcześniej graf, przy okazji wskazując moduł pamięci do przechowywania jego stanu
(`checkpointer=memory`)
"""
graph_config = {"configurable": {"thread_id": uuid.uuid4()}}
compiled_graph = graph_builder.compile(checkpointer=memory)
"""
Definiujemy prompt, czyli zapytanie do LLM-a.
Proszę zauważyć że, w zapytaniu wskazujamy również źródło danych do przeszukania (w tym przypadku stanowi ono nasza
baza wiedzy) oraz określamy oczekiwany format odpowiedzi (JSON).
Co ciekawe, w pliku bazy wiedzy celowo zaciemniliśmy kontekst wprowadzając do niego kilka nieistotnych informacji
po to, aby sprawdzić jak model poradzi sobie z utrudnieniami w dotarciu do informacji.
"""
prompt = {"messages": [
SystemMessage(content="Provide the temperature for all cities described in `knowledge-base`. Respond in JSON format without any additional text (JSON only without markdown).")
]}
"""
Wyświetlamy wiadomości z całego procesu - do momentu jego zatrzymania w oczekiwaniu na interakcję z człowiekiem
"""
for event in compiled_graph.stream(prompt, config=graph_config, stream_mode="values"):
stream_parser(event)
"""
Odbieramy odpowiedź od człowieka
"""
human_response_input = input()
"""
Ostatecznie, kontynuujemy wyświetlanie wiadomości po interakcji z człowiekiem
"""
for event in compiled_graph.stream(Command(resume=human_response_input), config=graph_config, stream_mode="updates"):
stream_parser(event)
Kompletny kod aplikacji znajdziesz na końcu tego artykułu.
Teraz przyjrzyjmy się natomiast wyjściu z aplikacji po jej uruchomieniu:
================================ System Message ================================
Provide the temperature for all cities described in `knowledge-base`. Respond in JSON format without any additional
text (JSON only without markdown).
================================== Ai Message ==================================
Tool Calls:
context_searcher (call_kZ7MnU6UjP2KlHd80P6bdok7)
Call ID: call_kZ7MnU6UjP2KlHd80P6bdok7
Args:
query: temperature
================================= Tool Message =================================
Name: context_searcher
Here are some words to confuse the LLM and create confusion in the context. The average temperature
confusion in the context. The average temperature in Bialystok is 27 C.
Meanwhile, Warsaw can boast an average of 22.5 C.
However, in Gdansk it can be 25 C. And here is some more content to confuse the model again and see
================================== Ai Message ==================================
Tool Calls:
context_searcher (call_L46mEnkzlfjfJu3DCZZwhhMF)
Call ID: call_L46mEnkzlfjfJu3DCZZwhhMF
Args:
query: Bialystok temperature
context_searcher (call_9TgcqpBqjp4PayXy1mP0DL7w)
Call ID: call_9TgcqpBqjp4PayXy1mP0DL7w
Args:
query: Warsaw temperature
context_searcher (call_2mWkttSW8tbo8sG25NpXCrYT)
Call ID: call_2mWkttSW8tbo8sG25NpXCrYT
Args:
query: Gdansk temperature
================================= Tool Message =================================
Name: context_searcher
Meanwhile, Warsaw can boast an average of 22.5 C.
Meanwhile, Warsaw can boast an average of 22.5 C.
Meanwhile, Warsaw can boast an average of 22.5 C.
Meanwhile, Warsaw can boast an average of 22.5 C.
================================== Ai Message ==================================
Tool Calls:
context_searcher (call_7h7SIsFlIvXJ7ZamsMN36J6g)
Call ID: call_7h7SIsFlIvXJ7ZamsMN36J6g
Args:
query: Bialystok
context_searcher (call_KzuJXj6S1MioIbnc7p9KBTH3)
Call ID: call_KzuJXj6S1MioIbnc7p9KBTH3
Args:
query: Warsaw
context_searcher (call_uYle26b7Dd1rrQgEq8WWP5t3)
Call ID: call_uYle26b7Dd1rrQgEq8WWP5t3
Args:
query: Gdansk
================================= Tool Message =================================
Name: context_searcher
Meanwhile, Warsaw can boast an average of 22.5 C.
Meanwhile, Warsaw can boast an average of 22.5 C.
Meanwhile, Warsaw can boast an average of 22.5 C.
Meanwhile, Warsaw can boast an average of 22.5 C.
================================== Ai Message ==================================
{
"Bialystok": "27 C",
"Warsaw": "22.5 C",
"Gdansk": "25 C"
}
>>> User interaction <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
{'question': 'Hi human! :) If the answer are correct? Type: `y` or `n`'}
y
>>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
Your answer: y
>>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
✅ Do something in approved path.
Co tu się dokładnie wydarzyło?
Po pierwsze, system użył zdefiniowanego promptu, celem uzyskania informacji o bieżącej temperaturze w poszczególnych miastach w Polsce (co warto zaznaczyć – LLM zrobił to bez konkretnego wskazania tychże miast).
Dodatkowo, LLM odnalazł te informacje, pomimo częściowego zaciemnienia danych i na końcu poprosił człowieka o zatwierdzenie poprawności odpowiedzi.
Podsumowanie
Jak widać na powyższym przykładzie, implementacja koncepcji Human-in-the-loop z użyciem LangGraph jest kwestią relatywnie prostą.
Co ważne, pozwala skutecznie łączyć możliwości LLM-ów z dodatkowym kontekstem (bazą wiedzy) oraz z niezbędną, ludzką kontrolą.
Powyższy tandem pozwala wznieść użyteczność aplikacji na niespotykany dotychczas poziom.
Kompletny kod aplikacji
Zawartość pliku .env (zmienne środowiskowe):
OPENAI_API_KEY=TWOJ_KLUCZ_OPEN_AI
ANONYMIZED_TELEMETRY=False
Zawartość pliku knowledge-base.txt (baza wiedzy):
Here are some words to confuse the LLM and create confusion in the context. The average temperature in Bialystok is 27 C.
However, in Gdansk it can be 25 C. And here is some more content to confuse the model again and see how it handles with this.
Meanwhile, Warsaw can boast an average of 22.5 C.
"""
Base Python libraries
"""
import os
import uuid
from pathlib import Path
from typing import Literal
from dotenv import load_dotenv
from IPython.display import Image, display
"""
Required LangGraph and LangChain libraries
"""
from langchain_core.messages import SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.types import Command, interrupt
"""
Context search libraries
"""
from langchain_community.vectorstores import Chroma
from langchain_community.vectorstores.utils import filter_complex_metadata
from langchain_unstructured import UnstructuredLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
"""
-------------
Configuration
-------------
Enable more details in LLM response (debug logs)
"""
detailed_model_response = False
"""
Load external environment variables contained project specific data (API keys etc.)
"""
load_dotenv(dotenv_path=Path(".env"))
"""
--------------------------------------------------------------------------------------------------------------------
Data source / Knowledge Base configuration
--------------------------------------------------------------------------------------------------------------------
"""
datasource = 'knowledge-base.txt'
"""
--------------------------------------------------------------------------------------------------------------------
Build memory
--------------------------------------------------------------------------------------------------------------------
"""
memory = MemorySaver()
"""
--------------------------------------------------------------------------------------------------------------------
Utils
--------------------------------------------------------------------------------------------------------------------
"""
def stream_parser(stream_message):
if "messages" in stream_message:
stream_message["messages"][-1].pretty_print()
if "__interrupt__" in stream_message:
print(">>> User interaction <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n")
print(stream_message["__interrupt__"][-1].value)
else:
if detailed_model_response:
print(stream_message)
print("\n")
"""
--------------------------------------------------------------------------------------------------------------------
Tools
--------------------------------------------------------------------------------------------------------------------
Tool required to search the relevant documents (like prepared knowledge-base)
"""
@tool
def context_searcher(query: str):
"""Search the relevant documents"""
loader = UnstructuredLoader(datasource)
document = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=50)
split_documents = text_splitter.split_documents(document)
filtered_documents = filter_complex_metadata(split_documents)
vectorstore = Chroma.from_documents(
documents=filtered_documents,
collection_name="knowledge-base",
embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
results = retriever.invoke(query)
return "\n".join([doc.page_content for doc in results])
"""
--------------------------------------------------------------------------------------------------------------------
AI Agent
--------------------------------------------------------------------------------------------------------------------
List of available tools
"""
tools = [context_searcher]
"""
Initialize the LLM model and attach tools
"""
openai_api_key = os.getenv("OPENAI_API_KEY")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=openai_api_key).bind_tools(tools)
"""
--------------------------------------------------------------------------------------------------------------------
Functions required by Human in the loop concept
--------------------------------------------------------------------------------------------------------------------
"""
def human_interaction(state: MessagesState) -> Command[Literal["human_approved", "human_rejected", END]]:
"""note: we not use the state in the current graph example - remember the state contains the LLM context"""
answer = interrupt(
{
"question": "Hi human! :) If the answer are correct? Type: `y` or `n`",
}
)
print(">>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n")
print("Your answer: ", answer, "\n\n")
if answer == "y":
return Command(goto="human_approved")
if answer == "n":
return Command(goto="human_rejected")
else:
print("Unsupported answer. Terminating...")
return Command(goto=END)
def human_approved(state: MessagesState) -> Command[END]:
print(">>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n")
print("✅ Do something in approved path.")
return Command(goto=END)
def human_rejected(state: MessagesState) -> Command[END]:
print(">>> Agent message <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n")
print("❌ Do something in rejected path.")
return Command(goto=END)
"""
--------------------------------------------------------------------------------------------------------------------
Build state graph workflow
--------------------------------------------------------------------------------------------------------------------
Get LLM response function
"""
def call_model(state: MessagesState):
messages = state['messages']
response = model.invoke(messages)
return {"messages": [response]}
"""
Conditional-edge function
"""
def should_continue(state: MessagesState) -> Literal["tools", "human_interaction"]:
messages = state['messages']
last_message = messages[-1]
"""If the model needs a tool, then call the "tools" node"""
if last_message.tool_calls:
return "tools"
return "human_interaction"
"""
Create a new State graph (workflow)
"""
graph_builder = StateGraph(MessagesState)
"""
Add an agent node
"""
graph_builder.add_node("agent", call_model)
"""
Add tools node
"""
graph_builder.add_node("tools", ToolNode(tools))
"""
Add human interaction nodes (HITL)
"""
graph_builder.add_node("human_interaction", human_interaction)
graph_builder.add_node("human_approved", human_approved)
graph_builder.add_node("human_rejected", human_rejected)
"""
Configure workflow conditions (edges)
"""
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges("agent", should_continue)
graph_builder.add_edge("tools", "agent")
"""
--------------------------------------------------------------------------------------------------------------------
Compile and run
--------------------------------------------------------------------------------------------------------------------
"""
graph_config = {"configurable": {"thread_id": uuid.uuid4()}}
compiled_graph = graph_builder.compile(checkpointer=memory)
"""
Streaming the model output
"""
prompt = {"messages": [
SystemMessage(content="Provide the temperature for all cities described in `knowledge-base`. Respond in JSON format without any additional text (JSON only without markdown).")
]}
for event in compiled_graph.stream(prompt, config=graph_config, stream_mode="values"):
stream_parser(event)
"""
Waiting for human response (if needed)
"""
human_response_input = input()
for event in compiled_graph.stream(Command(resume=human_response_input), config=graph_config, stream_mode="updates"):
stream_parser(event)
"""
Visualize your graph
"""
display(Image(compiled_graph.get_graph().draw_mermaid_png())) # works only in Jupiter notebooks
Zostaw komentarz