Практические советы: создание ИИ-ассистента для компании с использованием RAG, агентов и LangGraph

В следующих заметках описывается практический подход к созданию ИИ-ассистента для компании с использованием технологий RAG, Agents и LangGraph. Основное внимание уделяется контрактам, проверкам и шаблонам кода, а не мотивирующим формулировкам. При изучении обзора сначала запишите условия работы: необходимые входные данные, сигналы успешного выполнения и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Если какой-то шаг не сработает, причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций.

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}"
                )

Чек-лист операционной работы

Для чек-листа операционной работы необходимо заранее определить входные данные, ответственного за выполнение каждого шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность повторно выполнить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние системы.

Необходимо одновременно задокументировать успешный сценарий работы и сценарий восстановления. Повторные попытки, проверки со стороны человека и обработка неработоспособных сообщений являются неотъемлемой частью продукта, а не элементами, добавляемыми позже.

Указывайте конкретные фрагменты текста, на которые опирается ответ. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.

Пункт проверки после дорогостоящих шагов. Система возобновления работы не должна снова взимать плату за один и тот же вызов LLM, когда оператор пытается выполнить задачу в более позднем узле.

Фиксируйте версии зависимостей и записывайте хэш изображения, с использованием которого выполнялась демонстрация. Воспроизводимость важнее устного опыта сотрудников.

Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-версии к общим средам.

Перед переходом на более сложную версию стека заморозьте версии, сохраните эталонный текст для критического пути и убедитесь в наличии шагов для отката. В общих средах необходимы ограничения на частоту запросов, проверки принадлежности и четко определенный ответственный за обновление секретов. Лучше надежность, несмотря на ее простоту, чем креативные одноразовые демонстрации.

Примечание для 656e1cd26886: не храните ключи поставщика в репозитории, установите лимит токенов на каждую сессию и сохраняйте тексты рядом с фиксами для оценки, чтобы последующие замены моделей оставались сопоставимыми.

Для пункта усиления безопасности №0 необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии системы. Конфигурацию следует хранить отдельно от кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь кодовый граф.

Подробности усиления безопасности 0/780: измеряйте время выполнения, класс ошибок и расход токенов для данного пункта, затем принимайте решение о сохранении изменений на основе заранее определенного набора критериев, а не на основе устных оценок.

При работе над пунктом усиления безопасности №1 сначала запишите описание процесса: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Лучше использовать небольшие, тестируемые модули вместо обширных скриптов. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций.

Деталь усиления безопасности 1/780: измерьте время выполнения, класс ошибки и расход токенов для данного пункта, затем решите, следует ли сохранять изменение на основе определенного набора критериев, а не на основе устных замечаний.

Пункт усиления безопасности №2 будет работать наилучшим образом, если рассматривать его как объект с измеримыми показателями. Соберите один эталонный пример работы, один случай сбоя и запись о процедуре возврата к предыдущему состоянию перед расширением объема работ. Записывайте временные показатели и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.

Подробности усиления безопасности 2/780: измерьте время выполнения, класс ошибки и расход токенов для этой записи, затем решите, следует ли сохранить изменение на основе фиксированного набора вопросов, а не на основе единичных случаев.

Для записи об усилении безопасности 3 определите входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Документируйте как успешный путь, так и путь восстановления одновременно. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не последующими улучшениями.

Подробности усиления безопасности 3/780: измерьте время выполнения, класс ошибки и расход токенов для этой записи, затем решите, следует ли сохранить изменение на основе фиксированного набора вопросов, а не на основе единичных случаев.

При работе над пунктом усиления безопасности №4 сначала запишите контракт: необходимые входные данные, сигнал о успехе и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными данными. Дайте названия результатам работы, определите критерии успеха и не допускайте молчаливого частичного выполнения задач.

Подробности усиления безопасности 4/780: измерьте время выполнения, класс ошибки и расход токенов для данного пункта, затем решите, следует ли сохранять изменение на основе фиксированного набора критериев, а не на основе единичных примеров.

Пункт усиления безопасности №5 будет работать наилучшим образом, если рассматривать его как измеримую поверхность. Соберите один эталонный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять, не читая весь код.

Подробности усиления безопасности 5/780: измерьте время выполнения операции, класс ошибки и количество потраченных токенов для этой записи, затем решите, следует ли сохранить изменения на основе фиксированного набора критериев, а не на основе единичных примеров.

Для записи о усилении безопасности 6 определите входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Предпочтительнее использовать небольшие, проверяемые единицы кода вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на сложную взаимосвязь между компонентами.

Подробности усиления безопасности 6/780: измерьте время выполнения операции, класс ошибки и количество потраченных токенов для этой записи, затем решите, следует ли сохранить изменения на основе фиксированного набора критериев, а не на основе единичных примеров.

При работе над пунктом усиления безопасности №7 сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения, стоимость токенов или запросов. Очевидность затрат с самого начала предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.

Подробность усиления безопасности 7/780: измерьте время выполнения, класс ошибки и расход токенов для данного пункта, затем решите, следует ли сохранять изменение на основе фиксированного набора критериев, а не на основе устных замечаний.

Пункт усиления безопасности №8 будет работать наилучшим образом, если рассматривать его как измеримую составляющую. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

Подробности усиления безопасности 8/780: измерьте время выполнения, класс ошибки и расход токенов для этой записи, затем решите, следует ли сохранить изменения на основе фиксированного набора вопросов, а не на основе единичных примеров.

Для записи об усилении безопасности 9 определите входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия результатам работы, определите критерии успеха и откажитесь от молчаливого частичного завершения задачи.

Подробности усиления безопасности 9/780: измерьте время выполнения, класс ошибки и расход токенов для этой записи, затем решите, следует ли сохранить изменения на основе фиксированного набора вопросов, а не на основе единичных примеров.