Wskazówki praktyczne: Tworzenie asystenta AI dla firmy przy użyciu RAG, agentów i LangGraph

Poniższe notatki przedstawiają praktyczny plan działania dotyczący „Tworzenia asystenta AI dla firmy przy użyciu RAG, Agentów i LangGraph”. Nacisk kładziony jest na umowy, sprawdzenia oraz miejsca na kod do wstawienia, a nie na motywacyjne aspekty. Podczas analizy przeglądu 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. Wolimy 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.

rag-agent-project/
│
├── documents/
├── faiss_index/
│
├── .env
├── .gitignore
├── requirements.txt
│
├── config.py
├── ingest.py
├── rag.py
├── query_rewriter.py
├── reranker.py
├── tools.py
├── graph.py
├── main.py
└── app.py

pip install -r requirements.txt

.env

OPENAI_API_KEY=your_openai_api_key

import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

if not OPENAI_API_KEY:
    raise ValueError("OPEN_API_KEY is not set."
                     "Add it to yor .env file.")

LLM_MODEL = "gpt-4.1-mini"

EMBEDDING_MODEL = "text-embedding-3-small"

HUGGINGFACE_EMBEDDING = "sentence-transformers/all-MiniLM-L6-v2"
FAISS_PATH = "faiss_index"

DATABASE_PATH = "company_assistant.db"

documents/
├── employee_handbook.pdf
├── hr_policy.pdf
└── it_policy.pdf

from pathlib import Path

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings

from config import HUGGINGFACE_EMBEDDING , FAISS_PATH


DOCUMENTS_DIR = Path("documents")


def load_documents():

    all_documents = []

    pdf_files = list(DOCUMENTS_DIR.glob("*.pdf"))

    if not pdf_files:
        raise FileNotFoundError(
            "No pdf file found in documents/"
        )

    for pdf_file in pdf_files:

        print(f"Loading: {pdf_file}")

        loader = PyPDFLoader(str(pdf_file))
        documents = loader.load()
        for document in documents:
            document.metadata["source"] = pdf_file.name
            if "handbook" in pdf_file.name.lower():
                document.metadata["document_type"] = "handbook"

            elif "policy" in pdf_file.name.lower():

                document.metadata["document_type"] = "policy"

            else:

                document.metadata["document_type"] = "other"

            all_documents.append(document)

    return all_documents


def main():
    # 1. Load PDFs
    documents = load_documents()

    print(f"\nLoaded {len(documents)} pages")

    # 2. Split documents
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=100
    )

    chunks = text_splitter.split_documents(documents)

    print(f"Split into {len(chunks)} chunks.")

    # 3. Create embeddings
    embeddings = HuggingFaceEmbeddings(
        model_name=HUGGINGFACE_EMBEDDING
    )

    # 4. Create FAISS database
    vector_store = FAISS.from_documents(
        chunks,
        embeddings
    )


    # 5. Save FAISS database
    vector_store.save_local(FAISS_PATH)

    print(
        f"\nFAISS index saved to: {FAISS_PATH}"
    )


if __name__ == "__main__":
    main()

Loading: documents/employee_handbook.pdf
Loading: documents/hr_policy.pdf
Loading: documents/it_policy.pdf
Loaded 20 pages.
Created 60 chunks.
FAISS index created successfully!

Chunk 1
-------------------------
A B C D E F G H I J
              |
              | overlap
              v
Chunk 2
              E F G H I J K L M N

from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from config import(
    HUGGINGFACE_EMBEDDING,
    FAISS_PATH
)

embeddings = HuggingFaceEmbeddings(
        model_name=HUGGINGFACE_EMBEDDING
    )

vector_store = FAISS.load_local(
    FAISS_PATH,
    embeddings,
    allow_dangerous_deserialization=True
)

from rag import vector_store

from langchain_huggingface import (HuggingFaceEmbeddings,HuggingFacePipeline)
from transformers import pipeline

pipe = pipeline(
    "text-generation",
    model="Qwen/Qwen2.5-0.5B-Instruct",
    max_new_tokens=256,
    temperature=0.1
)

llm = HuggingFacePipeline(
    pipeline=pipe
)

def rewrite_query(question:str)-> str:
    prompt = f"""
Rewrite the user's question into a concise search query for company document database.

The query should contain important keywords that are likely to appear in company documents.

Do not answer the question.

User question:{question}

search query:
"""

    response = llm.invoke(prompt)

    return response

Chunk 1
Chunk 2
Chunk 3
Chunk 4
Chunk 5

from langchain_huggingface import (HuggingFaceEmbeddings,HuggingFacePipeline)
from transformers import pipeline

pipe = pipeline(
    "text-generation",
    model="Qwen/Qwen2.5-0.5B-Instruct",
    max_new_tokens=256,
    temperature=0.1
)

llm = HuggingFacePipeline(
    pipeline=pipe
)

def rerank_documents(query:str, documents:list,top_n: int = 3):

    scored_documents =[]

    for document in documents:
        prompt = f"""
You are evaluating document releveance.

Rate how relevant the document is to the user's question.

Return ONLY an intefer from 0 to 10.

0 = competely irrelevent
10 = extreamly relevent

Question: {query}

Document:{document.page_content}

"""
        response = llm.invoke(prompt)
        try:
            score = int(response.content.strip())
        except ValueError:
            score = 0

        scored_documents.append((score,document))

        #Highest score first
        scored_documents.sort(
            key=lambda item:item[0],
            reverse=True
        )

        return[
            document
            for score, document
            in scored_documents[:top_n]
        ]

from langchain.tools import tool
from rag import vector_store
from query_rewriter import(rewrite_query)

from reranker import (rerank_documents)


@tool
def company_document_search(query: str) -> str:
    """
    Search company PDF documents for information about

    company policies, employee benifits, leave, holidays, work from home,
    IT rules, working hours, and othher internal information.

    """
    #1. Rewrite Query
    imporved_query = rewrite_query(query)
    print(f"\n[Rewritten Query]"
          f"\n{imporoved_query}")

    #2.  Vector serach
    documents =  (vector_store.similarity_search(
        imporved_query,
        k=5
    ))

    if not documents:
        return(
            "No relevent information"
            "was found in the company documents"
        )

    #Reranking
    documents = rerank_documents(
        imporved_query, documents,top_n=3
    )

    #4.Formate context
    result = []
    for document in documents:
        source = document.metadata.get("source","Unknown")
        page = document.metadata.get("page","Unknown")

        #PyPDFLoader user zero-based pages
        if isinstance(page, int):
            page = page+1

        results.append(f"""
SOURCE:{source}
PAGE:{page}

CONTENT: {document.page_content}
""")
        return "\n".join(result)


@tool
def calculator(expression: str) -> str:
    """
    Calculate a mathematical expression.
    """
    try:

       # result = eval(expression)
       # for learning only
       # Do not use eval() in production

        allowed = ("0123456789""+-*/(). ")
        if not all(char in allowed for char in expression):
            return ("Invalid mathmatical expression")

        result = eval(expression,
                      {"__builtins__":{}},
                      {}
                      )

        return str(result)
    except Exception:
        return "Unable to calculate the expression."

from langchain_ollama import ChatOllama

from langgraph.graph import (
    StateGraph,
    MessagesState,
    START
)

from langgraph.prebuilt import (
    ToolNode,
    tools_condition
)

from langgraph.checkpoint.memory import InMemorySaver

from tools import (
    company_document_search,
    calculator
)

# Local Qwen model
llm = ChatOllama(
    model="qwen2.5:3b",
    temperature=0
)

# Tools
tools = [
    company_document_search,
    calculator
]


llm_with_tools = llm.bind_tools(tools)

# Agent

def agent_node(state: MessagesState):

    response = llm_with_tools.invoke(
        state["messages"]
    )

    return {
        "messages": [response]
    }

# Graph
graph_builder = StateGraph(MessagesState)


graph_builder.add_node(
    "agent",
    agent_node
)


graph_builder.add_node(
    "tools",
    ToolNode(tools)
)

# START → Agent
graph_builder.add_edge(
    START,
    "agent"
)

# Agent → Tool OR END
graph_builder.add_conditional_edges(
    "agent",
    tools_condition
)

# Tool → Agent
graph_builder.add_edge(
    "tools",
    "agent"
)

# Checkpointer
checkpointer = InMemorySaver()

# Compile

graph = graph_builder.compile(
    checkpointer=checkpointer
)

SOURCE: hr_policy.pdf
PAGE: 4

Source: hr_policy.pdf, page 4.

from langchain_core.messages import HumanMessage, SystemMessage
from graph import graph

SYSTEM_PROMPT="""
You are an intelligent company assistant.

You answer question using company documents and avalable tools.

Rules:
1. Use company_document_search whenever the user asks about company-specific information.

2. Never invent company policies,  benifits, rules, or  procedures.

3. Use the calculator when mathmatical calculation are required.

4. When company documents are used, mention the source document and page.

5. If the company documents do not contain the requested information, clearly say
that you could not find it.

6. Use previous conversation context to undersatnd follow-up questions.

7. Give concise and clear answer
"""

def chat(
        question:str,
        thread_id: str):
    config = {
        "configurable":{"thread_id":thread_id}
    }

    #Check whether thhis conversation already exists
    state = graph.get_state(config)

    if not state.values:
        messages = [
            SystemMessage(content=SYSTEM_PROMPT),
            HumanMessage(content=question)
        ]
    else:
        messages=[
            HumanMessage(content=question)]

    result = graph.invoke({
        "messages":messages},
        config=config
    )
    return result["messages"][-1].content

def main():
    print("===========================================")
    print("CoOMPANY AI ASSISTANT")
    print("===========================================")
    print("\nType 'exist' to quit.")
    thread_id = input("\nEnter conversation ID:")

    while True:
        question = input("\nYou: ")

        if question.lower() in [
            "exit",
            "quit"
        ]:
            break

        try:
            answer = chat(question,thread_id)
            print(f"\nAI: {answer}")

        except Exception as e:
            print(f"\nError: {e}")

if __name__ == "__main__":
    main()

import streamlit as st
from langchain_core.messages import (
    HumanMessage,
    AIMessage,
    SystemMessage
)

from graph import graph

# Page configuration
st.set_page_config(
    page_title="Company AI Assistant",
    page_icon="🤖",
    layout="centered"
)

# System prompt
SYSTEM_PROMPT = """
You are an intelligent company assistant.

You answer questions using company documents
and available tools.

Rules:

1. Use company_document_search whenever
   the user asks about company-specific
   information.

2. Never invent company policies,
   benefits, rules, or procedures.

3. Use the calculator when mathematical
   calculations are required.

4. When company documents are used,
   mention the source document and page.

5. If the company documents do not contain
   the requested information, clearly say
   that you could not find it.

6. Use previous conversation context to
   understand follow-up questions.

7. Give concise and clear answers.
"""

# Sidebar
st.sidebar.title("⚙️ Settings")

thread_id = st.sidebar.text_input(
    "Conversation ID",
    value="streamlit-user"
)

st.sidebar.markdown(
    """
### About

This application demonstrates:

- RAG
- AI Agents
- LangGraph
- FAISS
- Query rewriting
- Reranking
- Conversation memory
- Tool calling
"""
)

# Main UI
st.title("🏢 Company AI Assistant")

st.write(
    "Ask questions about company policies, "
    "benefits, leave, IT rules, and more."
)

# Chat history in Streamlit

if "chat_history" not in st.session_state:
    st.session_state.chat_history = []

# Display previous messages
for message in st.session_state.chat_history:

    with st.chat_message(
        message["role"]
    ):

        st.markdown(
            message["content"]
        )


# User input
question = st.chat_input(
    "Ask something about the company..."
)


if question:
    # Display user question
    with st.chat_message("user"):

        st.markdown(question)


    st.session_state.chat_history.append(
        {
            "role": "user",
            "content": question
        }
    )

    # Build messages
    config = {
        "configurable": {
            "thread_id": thread_id
        }
    }

    # Check whether this thread
    # already has state
    state = graph.get_state(
        config
    )
    if not state.values:

        messages = [
            SystemMessage(
                content=SYSTEM_PROMPT
            ),
            HumanMessage(
                content=question
            )
        ]

    else:
        messages = [
            HumanMessage(
                content=question
            )
        ]

    # Call LangGraph
    with st.chat_message("assistant"):
        with st.spinner(
            "Thinking..."
        ):
            try:
                result = graph.invoke(
                    {
                        "messages": messages
                    },
                    config=config
                )
                answer = result[
                    "messages"
                ][-1].content
                st.markdown(answer)

                # Save answer
                st.session_state.chat_history.append(
                    {
                        "role": "assistant",
                        "content": answer
                    }
                )
            except Exception as e:

                st.error(
                    f"Something went wrong: {e}"
                )

Lista kontrolna operacyjna

Przy tworzeniu listy kontrolnej operacyjnej należy zdefiniować dane wejściowe, osobę odpowiedzialną za daną czynność oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie wykonać daną czynność na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu.

Należy udokumentować zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponownych działań, mechanizmy ludzkiej kontroli oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.

Należy podać fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od braku informacji w indeksie.

Punkt kontrolny po kosztownych krokach. System powinien unikać ponownego naliczania opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.

Zamrożone wersje zależności oraz zapisany digest obrazu, który uruchomił demonstrację – to klucz do powtarzalności. Powtarzalność jest lepsza od lokalnej wiedzy specjalistów.

Zapisuj czas trwania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się od demonstracji do wspólnych środowisk.

Zanim wdrożymy całą architekturę, zamroźmy wersje oprogramowania, utwórzmy idealny zapis transkrypcji dla kluczowych ścieżek działania i potwierdźmy kroki odwracające zmiany. Wspólne środowiska wymagają ograniczeń szybkości, weryfikacji dostępu oraz jasno określonego właściciela odpowiedzialnego za rotację haseł. Lepiej mieć nudną, ale niezawodną architekturę niż genialne, jednorazowe demonstracje.

Uwaga dotycząca wersji 656e1cd26886: nie umieszczaj kluczy dostawcy w repozytorium, ustaw górny limit tokenów na sesję oraz przechowuj transkrypcje obok plików testowych, aby późniejsze zmiany modeli pozostały porównywalne.

Dla uwagi dotyczącej wzmocnienia bezpieczeństwa nr 0 należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić, nie musząc czytać całej struktury.

Szczegół wzmocnienia bezpieczeństwa 0/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej uwagi, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie informacji anegdotycznych.

Gdy pracujesz nad ulepszeniem nr 1, 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. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces.

Szczegół ulepszenia 1/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tego punktu, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie osobistych obserwacji.

Ulepszenie nr 2 działa najlepiej, gdy traktuje się je jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przykład działania, jeden przypadek niepowodzenia oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres prac. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do wspólnych środowisk.

Szczegół wzmocnienia 2/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Dla notatki dotyczącej wzmocnienia 3 zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie wykonać ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zdokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponowne, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie element późniejszej dopracowywania.

Szczegół wzmocnienia 3/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Gdy pracujesz nad uwagą dotyczącą wzmocnienia bezpieczeństwa nr 4, 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ć przypadki częściowego ukończenia bez żadnych informacji.

Szczegół wzmocnienia bezpieczeństwa 4/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej uwagi, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie informacji anegdotycznych.

Uwaga dotycząca wzmocnienia bezpieczeństwa nr 5 sprawdza się najlepiej, gdy jest traktowana jako mierzalna powierzchnia do analizy. Zapisz jeden idealny przykład działania, jeden przypadek niepowodzenia oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres prac. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury.

Szczegół wzmocnienia 5/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Dla notatki dotyczącej wzmocnienia nr 6 zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces.

Szczegół wzmocnienia 6/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Gdy pracujesz nad uwagą nr 7 dotyczącą wzmocnienia bezpieczeństwa, najpierw zapisz specyfikację: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego awarii. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Obok wyników funkcjonalnych zapisz czas wykonywania oraz koszt tokena lub zapytania. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.

Szczegół nr 7/780 dotyczący wzmocnienia bezpieczeństwa: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej uwagi, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie informacji anegdotycznych.

Uwaga nr 8 dotycząca wzmocnienia bezpieczeństwa działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię do analizy. Zanim rozszerzysz zakres, zapisz jeden idealny przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Dokumentuj razem ścieżkę prawidłowego działania oraz ścieżkę przywracania stanu. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.

Szczegóły wzmocnienia bezpieczeństwa 8/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Dla notatki dotyczącej wzmocnienia bezpieczeństwa nr 9 zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie wykonać ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania.

Szczegóły wzmocnienia bezpieczeństwa 9/780: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.