Sztuczna inteligencja agentowa z LangChain, część 1
Część 1: Sztuczna inteligencja agentowa z LangChain – jak model językowy otrzymuje narzędzia, pamięć i pętlę rozumowania, jak RAG wpisuje się w funkcjonowanie agenta oraz jakie testy należy przeprowadzić przed jego wdrożeniem.
Źródła: Agentic AI with LangChain — Part 1: Building A Semantic RAG Pipeline
To przewodnik pokazuje, jak odtworzyć proces od surowców do działającego systemu w przypadku: Agentic AI z LangChain — Część 1: Budowa semantycznego pipeline RAG. Skupiamy się na krokach operacyjnych, wyraźnych sprawdzeniach oraz kodzie, który można bez problemu wdrożyć do repozytorium, bez konieczności domyślania się intencji. Aby uzyskać ogólny obraz, zdefiniuj wprowadzenia, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności domyślania się ukrytego stanu. Zdokumentuj zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa błędów stanowią część produktu, a nie elementy dodawane później.
Stack technologiczny
Gdy pracujesz nad technologią, najpierw zapisz specyfikację: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolę małe, testowalne jednostki niż rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
1. Konfiguracje systemu
Gdy pracujesz nad rozdziałem 1. Konfiguracje systemu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadkowe, częściowe ukończenie zadań. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabą efektywność wyszukiwania.
OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>
class BaseConfig(BaseSettings):
OPENAI_API_KEY: Optional[str]
PINECONE_API_KEY: Optional[str]
TAVILY_API_KEY: Optional[str]
"""Loads the dotenv file. Including this is necessary to get
pydantic to load a .env file."""
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
2. Struktura projektu
Gdy pracujesz nad punktem 2. Struktura projektu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Zmierz stopień odzyskiwania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabą skuteczność wyszukiwania.
project/
├── rag_process/
│ ├── __init__.py
│ ├── extractor.py
| └── service.py
│
├── chains.py # llm chains
├── nodes.py # python functions utlizing llm chains
|── schemas.py # pydantic data models
|── upload.py # python functions to upload files asynchronously
├── main.py # fastAPI app
├── config.py # pydantic BaseConfig
│── dependencies.py # dependency inject to fastAPI endpoint function
├── test_chains.py # pytests
├── .env # environment variable definition
├── client.py # streamlit client
├── .gitignore
└── requirements.txt # package requirements
Gdy pracujesz nad punktem 2. Struktura projektu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrolne etapy przeprowadzane przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później.
2.1. Łańcuchy LLM
2.1. Łańcuchy LLM działają najlepiej, gdy traktuje się je jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Wolij małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Określ budżet tokenów na jeden ruch i na jedną sesję. Narzędzia typu agentic intensywnie rozszerzają kontekst; sztywne limity zapobiegają temu, by demonstracje przerodziły się w nieoczekiwane rachunki.
2.2. Symantic RAG
2.2. Symantic RAG funkcjonuje najlepiej, gdy traktuje się je jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Oddziel zasadę dzielenia na fragmenty od zasady wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisania drugiej, gdy zmieniają się metryki jakości.
2.3. Backend FastAPI
2.3. FastAPI Backend funkcjonuje najlepiej, gdy jest traktowany jako zmienna, którą można zmierzyć. Zanim rozszerzy się zakres, należy zapisać jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Obok wyników funkcjonalnych należy odnotować czas wykonywania operacji oraz koszt tokenów lub zapytań. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy aplikacja przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Należy oddzielić zasadę dzielenia na fragmenty od zasady pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości. 2.3. FastAPI Backend funkcjonuje najlepiej, gdy jest traktowany jako zmienna, którą można zmierzyć. Zanim rozszerzy się zakres, należy zapisać jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Należy udokumentować zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponownych wysyłek, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
2.4. Klient Streamlit
Dla wersji 2.4 klienta Streamlit należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Należy podawać konkretne fragmenty tekstu, które stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od braku informacji w bazie danych.
3. Łańcuchy LLM z LCEL
Dla łańcuchów LLM 3. z LCEL należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Przy następnym kroku będącym kodem lub wywołaniem narzędzia preferuj strukturyzowane wyniki z walidacją schematu nad tekstem w formie swobodnej.
3.1. Język wyrażeń LangChain (LCEL)
Dla 3.1. LangChain Expression Language (LCEL) należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Należy podawać fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie mogą odróżnić halucynacji od luki w indeksowaniu. Dla 3.1. LangChain Expression Language (LCEL) należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno ścieżkę pomyślnego działania, jak i ścieżkę naprawczą. Próby ponownych działań, mechanizmy ludzkiej kontroli oraz obsługa wiadomości błędowych stanowią część produktu, a nie coś dodanego później.
r polski.class GradeDocuments(BaseModel):
"""Binary score for relevance check on retrieved documents."""
binary_score: bool = Field(
description="Documents are relevant to the question, 'yes' or 'no'"
)
async def retrieval_grader(document: Document, question: str):
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)
system = """You are a grader assessing relevance of a retrieved document to a user question. \n
If the document contains keyword(s) or semantic meaning related to the question, grade it as relevant.\n
Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question. \n
if the document is relevant, give it a score 'yes', otherwise give it a score 'no'
"""
grade_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("human", "Retrieved document: \n\n {document} User question: {question}"),
]
)
retrieval_grader = grade_prompt | structured_llm_grader
grade = await retrieval_grader.ainvoke({"document": document, "question": question})
return grade.binary_score
4. Semantic RAG z PineCone
4.1. Proces przetwarzania danych
4.2. Proces wyszukiwania
4.3. Klasa VectorService
4.3.1. Inicjalizacja
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
class VectorService:
def __init__(self, index_name: str = "knowledgebase"):
self.index_name = index_name
# Embedding model
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small", api_key=api_key)
# Pinecone vectorstore wrapper
self.vectorstore = PineconeVectorStore(
index_name=index_name,
embedding=self.embeddings,
pinecone_api_key=pinecone_key,
)
# Retriever
self.retriever = self.vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3}
)
# Text splitter (LangChain-native)
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=50,
separators=["\n\n", "\n", ".", " ", ""]
)
4.3.2. Przechowywanie treści pliku
async def store_file_content_in_db(self, filepath: str) -> None:
"""
Read the uploaded text file, split the file, embed chunks and save chunks to vector database
:param filepath: filepath to load to vectordb
:return: None
"""
logger.debug(f"Loading file: {filepath}")
# Load entire file (safe for most RAG use cases)
async with aiofiles.open(filepath, "r", encoding="utf-8") as f:
raw_text = await f.read()
# Minimal cleaning (preserve structure)
raw_text = raw_text.replace("\r", "")
# Split into LangChain Documents
docs = self.splitter.create_documents(
[raw_text],
metadatas=[{"source": os.path.basename(filepath)}]
)
logger.debug(f"Storing {len(docs)} chunks into Pinecone")
# Ingest into Pinecone
PineconeVectorStore.from_documents(
documents=docs,
embedding=self.embeddings,
index_name=self.index_name,
pinecone_api_key=pinecone_key,
)
4.3.3. Wyszukiwanie dokumentów
async def search_documents(self, query: str) -> list[Document]:
logger.debug(f"Searching for: {query}")
return self.retriever.invoke(query)
5. Backend FastAPI
5.1. Wgracz plików
async def save_file(file: UploadFile) -> str:
"""
accept a UploadFile from FastAPI and save the file in specified location
:param file: UploadFile specified by users from fastAPI interface
:return: path of the saved file
"""
await makedirs("uploads", exist_ok=True)
filepath = os.path.join("uploads", file.filename)
async with aiofiles.open(filepath, "wb") as f:
while chunk := await file.read(DEFAULT_CHUNK_SIZE):
await f.write(chunk)
return filepath
5.2. Modele żądania/odpowiedzi
class RAGRequest(BaseModel):
prompt: str
class RAGResponse(BaseModel):
answer: str
num_original_documents: int
5.3. Koniec punktu generowania odpowiedzi
# dependencies.py
from fastapi import Body, HTTPException
from schemas import RAGRequest
from nodes import generate
async def get_generation(body: RAGRequest=Body(...)) -> dict:
try:
generation = await generate(body.prompt)
return {
"answer": generation.get("generation", ""),
"num_original_documents": generation.get("num_original_documents", 0)
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# main.py
@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(generation: dict = Depends(get_generation)) -> RAGResponse:
return RAGResponse(**generation)
5.4. Przetwarzanie w tle
@app.post("/upload")
async def file_upload_controller(
file: Annotated[UploadFile, File(description="Uploaded PDF documents")],
bg_text_processor: BackgroundTasks,
):
if file.content_type != "application/pdf":
raise HTTPException(
detail=f"Only uploading PDF documents are supported",
status_code=status.HTTP_400_BAD_REQUEST,
)
try:
filepath = await save_file(file)
bg_text_processor.add_task(pdf_text_extractor, filepath)
bg_text_processor.add_task(
vector_service.store_file_content_in_db,
filepath.replace("pdf", "txt")
)
except Exception as e:
raise HTTPException(
detail=f"An error occurred while saving file - Error: {e}",
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
return {"filename": file.filename, "message": "File uploaded successfully"}
6. Frontend Streamlit
6.1. Wgranie pliku
st.write("Upload a file to FastAPI")
file = st.file_uploader("Choose a file", type=["pdf"])
if st.button("Submit"):
if file is not None:
files = {"file": (file.name, file, file.type)}
response = requests.post("http://localhost:8000/upload", files=files)
st.write(response.text)
else:
st.write("No file uploaded.")
6.2. Historia rozmowy
if "messages" not in st.session_state:
st.session_state.messages = []
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
6.3. Proces rozmowy
if prompt := st.chat_input("Write your prompt in this input field"):
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.text(prompt)
question = prompt
response = requests.post(
f"http://localhost:8000/generate_text",
json={"prompt": question}
)
response.raise_for_status()
response_json = response.json()
answer = response_json["answer"]
st.session_state.messages.append({"role": "assistant", "content": answer})
with st.chat_message("assistant"):
st.markdown(answer)
7. Tworzenie streszczenia
Lista kontrolna operacyjna
Literatura pokrewna
- Praktyczne notatki: Część 3: Expert RAG — HyDE, Self-Querying i agentic workflows — Szczegółowy przewodnik po Praktycznych notatkach: Część 3: Expert RAG — HyDE, Self-Querying i agentic workflows: kontrakty, sprawdzenia oraz gotowe fragmenty kodu dla zespołów tworzących systemy RAG.
- Praktyczne notatki: Observable Voice RAG Agents: Śledzenie działań agentów RAG od początku do końca za pomocą MLflow i Qdrant — Szczegółowy przewodnik po Praktycznych notatkach: Observable Voice RAG Agents: Śledzenie działań agentów RAG od początku do końca za pomocą MLflow i Qdrant: kontrakty, sprawdzenia oraz gotowe fragmenty kodu dla zespołów.